static String write(Span span, boolean searchEnabled, ByteBufOutputStream sink) {
    int startIndex = sink.buffer().writerIndex();
    try (JsonGenerator writer = JsonSerializers.jsonGenerator(sink)) {
      writer.writeStartObject();
      if (searchEnabled) addSearchFields(span, writer);
      writer.writeStringField("traceId", span.traceId());
      if (span.parentId() != null) writer.writeStringField("parentId", span.parentId());
      writer.writeStringField("id", span.id());
      if (span.kind() != null) writer.writeStringField("kind", span.kind().toString());
      if (span.name() != null) writer.writeStringField("name", span.name());
      if (span.timestampAsLong() != 0L) {
        writer.writeNumberField("timestamp", span.timestampAsLong());
      }
      if (span.durationAsLong() != 0L) writer.writeNumberField("duration", span.durationAsLong());
      if (span.localEndpoint() != null && !EMPTY_ENDPOINT.equals(span.localEndpoint())) {
        writer.writeFieldName("localEndpoint");
        write(span.localEndpoint(), writer);
      }
      if (span.remoteEndpoint() != null && !EMPTY_ENDPOINT.equals(span.remoteEndpoint())) {
        writer.writeFieldName("remoteEndpoint");
        write(span.remoteEndpoint(), writer);
      }
      if (!span.annotations().isEmpty()) {
        writer.writeArrayFieldStart("annotations");
        for (int i = 0, length = span.annotations().size(); i < length; ) {
          write(span.annotations().get(i++), writer);
        }
        writer.writeEndArray();
      }
      if (!span.tags().isEmpty()) {
        writer.writeObjectFieldStart("tags");
        Iterator<Map.Entry<String, String>> tags = span.tags().entrySet().iterator();
        while (tags.hasNext()) write(tags.next(), writer);
        writer.writeEndObject();
      }
      if (Boolean.TRUE.equals(span.debug())) writer.writeBooleanField("debug", true);
      if (Boolean.TRUE.equals(span.shared())) writer.writeBooleanField("shared", true);
      writer.writeEndObject();
    } catch (IOException e) {
      throw new AssertionError(e); // No I/O writing to a Buffer.
    }

    // get a slice representing the document we just wrote so that we can make a content hash
    ByteBuf slice = sink.buffer().slice(startIndex, sink.buffer().writerIndex() - startIndex);

    return span.traceId() + '-' + md5(slice);
  }
In order to allow systems like Kibana to search by timestamp, we add a field "timestamp_millis"
when storing. The cheapest way to do this without changing the codec is prefixing it to the
json. For example. {"traceId":"... becomes {"timestamp_millis":12345,"traceId":"...

<p>Tags are stored as a dictionary. Since some tag names will include inconsistent number of
dots (ex "error" and perhaps "error.message"), we cannot index them naturally with
elasticsearch. Instead, we add an index-only (non-source) field of {@code _q} which includes
valid search queries. For example, the tag {@code error -> 500} results in {@code
"_q":["error", "error=500"]}. This matches the input query syntax, and can be checked manually
with curl.

<p>Ex {@code curl -s localhost:9200/zipkin:span-2017-08-11/_search?q=_q:error=500}

@param searchEnabled encodes timestamp_millis and _q when non-empty
zipkin-storage/elasticsearch/src/main/java/zipkin2/elasticsearch/internal/BulkIndexWriter.java