OpenTracing с Kafka Streams - как?

Question

Я пытаюсь интегрировать трассировку Jaeger в K-Streams. Я планировал добавить трассировку к нескольким из моих наиболее важных конвейеров и задавался вопросом, что будет хорошим способом передачи traceid с одной круговой линии на другую?

Вот что у меня получилось: в начале конвейера потоковой обработки я запускаю серверный промежуток и сохраняю traceid в хранилище состояний. Позже в конвейере преобразования я получаю доступ к хранилищу состояний и получаю трассировку из метода transform(). Это хороший способ обработки трассировки в потоковой обработке?

input
  .mapValues(toSomethingThatMyAppUnderstands)      
  .mapValues(this::startStreamTrace)
  .filter((k, v) -> v.isPresent())            
  .mapValues(Optional::get)                   
  .mapValues(doSomethingHereWith)       
  .flatMapValues(doSomethingElse)       
  .filter((k, v) -> isInterestingEvent(v))    
  .transform(() -> new TransformerWithTracing<SomeObjectA, SomeObjectB>(IN_MEM_STORE_NAME, someFunction), IN_MEM_STORE_NAME)
  .flatMapValues(c -> c)
  .to(outTopic, Produced.with(Serdes.String(), new EventSerde()));



public class TransformerWithTracing<V, VR> implements Transformer<String, V, KeyValue<String, VR>> {

  final Function valueAction;
  final String storeId;
  private KeyValueStore<String, String> traceIdStore;

  public TransformerWithTracing(String storeId, Function valueAction) {
    this.storeId = storeId;
    this.valueAction = valueAction;
  }

  @Override
  public void init(ProcessorContext context) {
   // KeyValueStore store = ((KeyValueStore<String, String>) context.getStateStore(storeId));
    InMemoryKeyValueStore inMemoryKeyValueStore = (InMemoryKeyValueStore) store;
    this.traceIdStore = store;
  }

  @Override
  public KeyValue<String, VR> transform(String key, V value) {
    System.out.println(traceIdStore.get(key));

    // BuildTraceHeader 
    try(Scope scope = serviceTracer.startServerSpan(traceHeader, "Converting to Enterprise Event")) {
      return KeyValue.pair(key, (VR) valueAction.apply(value));
    }
  }

  @Override
  public KeyValue<String, VR> punctuate(long timestamp) {
    return null;
  }

  @Override
  public void close() {
//    if (streamId != null)   traceIdStore.delete(streamId);
  }

}

Answer 1

В этом репозитории на молнии / смелых @jeqo есть похожие идеи.

https://github.com/jeqo/brave/tree/kafka-streams-processor/instrumentation/kafka-streams

Похоже, что в репозитории opentracing-contrib есть что-то доступное, но, похоже, только на уровне следов производителей / потребителей.

https://github.com/opentracing-contrib/java-kafka-client/tree/master/opentracing-kafka-streams

• Ленни