rx-java - Hystrix Observables で traceIds を渡す方法は?

Question

複数のサービスがあり、Hystrix の HystrixObservableCommand を使用して他のサービスを呼び出すサービスもあれば、HystrixCommand を使用するサービスもあります。呼び出し元のサービスから HystrixObservableCommand の Observables に traceIds を渡し、フォールバックが呼び出された場合にそれらを渡すにはどうすればよいですか?

すべてのサービスは grpc-java を使用しています。

私が持っているサンプルコード:

WorldCommand worldCommand = new WorldCommand(greeterRequest, worldServiceStub);
        String messageFromWorldService = "";
        String idFromWorldService = "";
        try {

            Greeter.GreeterReply greeterReply = worldCommand.construct().toBlocking().toFuture().get();
            messageFromWorldService = greeterReply.getMessage();
            idFromWorldService = greeterReply.getId();
            logger.info("Response from WorldService  -- {}, id = {}", messageFromWorldService, idFromWorldService);
        } catch (StatusRuntimeException | InterruptedException | ExecutionException e) {
            logger.warn("Exception when calling WorldService\n" +  e);
        }

WorldCommand.java

public class WorldCommand extends HystrixObservableCommand<Greeter.GreeterReply> {

    private static final Logger logger = LoggerFactory.getLogger(WorldCommand.class.getName());

    private final Greeter.GreeterRequest greeterRequest;
    private final WorldServiceGrpc.WorldServiceStub worldServiceStub;

    public WorldCommand(Greeter.GreeterRequest greeterRequest, WorldServiceGrpc.WorldServiceStub worldServiceStub) {
        super(HystrixCommandGroupKey.Factory.asKey("WorldService"));
        this.greeterRequest = greeterRequest;
        this.worldServiceStub = worldServiceStub;
    }

    @Override
    protected Observable<Greeter.GreeterReply> construct() {
        Context context = Context.current();
        return Observable.create(new Observable.OnSubscribe<Greeter.GreeterReply>() {
            @Override
            public void call(Subscriber<? super Greeter.GreeterReply> observer) {
                logger.info("In WorldCommand");
                if (!observer.isUnsubscribed()) {
                    //pass on the context, if you want only certain headers to pass on then create a new Context and attach it.
                    context.attach();
                    logger.info("In WorldCommand after attach");
                    worldServiceStub.greetWithHelloOrWorld(greeterRequest, new StreamObserver<Greeter.GreeterReply>() {
                        @Override
                        public void onNext(Greeter.GreeterReply greeterReply) {
                            logger.info("Response from WorldService  -- {}, id = {}", greeterReply.getMessage(), greeterReply.getId());
                            observer.onNext(greeterReply);
                            observer.onCompleted();
                        }

                        @Override
                        public void onError(Throwable t) {
                            logger.info("Exception from WorldService  -- {}", t);
                        }

                        @Override
                        public void onCompleted() {

                        }
                    });
                }
            }
        } ).subscribeOn(Schedulers.io());
    }

    @Override
    protected Observable<Greeter.GreeterReply> resumeWithFallback() {
        logger.info("Response from fallback");
        Greeter.GreeterReply greeterReply = Greeter.GreeterReply.newBuilder().setMessage("teammate").setId("-1").build();
        return Observable.just(greeterReply);
    }

Zipkin grpc トレースと MDCCurrentTraceContext を使用して、ログに traceId と spanId を出力しています。

WorldCommand の両方のログエントリは、トレース ID とスパン ID を出力しません。これらは RxIoScheduler スレッドで呼び出されます。

編集

Mike の提案に従って ConcurrencyStrategy を追加しました。

public class CustomHystrixConcurrencyStrategy extends HystrixConcurrencyStrategy {

    private static final Logger log = LoggerFactory.getLogger(CustomHystrixConcurrencyStrategy.class);

    public <T> Callable<T> wrapCallable(Callable<T> callable){
        log.info("In CustomHystrixConcurrencyStrategy: callable="+ callable.toString());
        return new ContextCallable<>(callable);
    }
}

HelloService は World と Team の 2 つのサービスを呼び出します。WorldCommand は HystrixObservableCommand であり、TeamCommand は HystrixCommand です。

logger.info("In the HelloService:greetWithHelloWorld");
Greeter.GreeterRequest greeterRequest = Greeter.GreeterRequest.newBuilder().setId(request.getId()).build();

//Call WorldService
ManagedChannel worldChannel = getChannel("localhost:8081", "helloService-world-client");
//Async stub instead of blockingStub
WorldServiceGrpc.WorldServiceStub worldServiceStub = WorldServiceGrpc.newStub(worldChannel);

WorldCommand worldCommand = new WorldCommand(greeterRequest, worldServiceStub);
String messageFromWorldService = "";
String idFromWorldService = "";
try {

    Greeter.GreeterReply greeterReply = worldCommand.construct().toBlocking().toFuture().get();
    messageFromWorldService = greeterReply.getMessage();
    idFromWorldService = greeterReply.getId();
    logger.info("Response from WorldService  -- {}, id = {}", messageFromWorldService, idFromWorldService);
} catch (StatusRuntimeException | InterruptedException | ExecutionException e) {
    logger.warn("Exception when calling WorldService\n" +  e);
}

//Call TeamService
ManagedChannel teamChannel = getChannel("localhost:8082", "helloService-team-client");
TeamServiceGrpc.TeamServiceBlockingStub teamServiceStub = TeamServiceGrpc.newBlockingStub(teamChannel);
TeamCommand teamCommand = new TeamCommand(greeterRequest, teamServiceStub);

String messageFromTeamService = "";
String idFromTeamService = "";
try {
    Greeter.GreeterReply greeterReply = teamCommand.construct().toBlocking().toFuture().get();
    messageFromTeamService = greeterReply.getMessage();
    idFromTeamService = greeterReply.getId();
    logger.info("Response from TeamService  -- {}, id = {}", messageFromTeamService, idFromTeamService);
} catch (StatusRuntimeException | InterruptedException | ExecutionException e) {
    logger.warn("Exception when calling TeamService\n" +  e);
}

assert(idFromWorldService.equals(idFromTeamService));
Greeter.GreeterReply greeterReply = Greeter.GreeterReply.newBuilder().setMessage("Hello" + messageFromWorldService + " from " + messageFromTeamService).setId(idFromWorldService).build();
responseObserver.onNext(greeterReply);
responseObserver.onCompleted();

PreservableContext クラス

public class PreservableContexts {

    //private final TraceContext traceContext;
    private static final Logger logger = LoggerFactory.getLogger(PreservableContexts.class.getName());

    public PreservableContexts() {
        logger.info("Creating new PreservableContexts");
        //this.traceContext = TraceContextHolder.getContext();
    }

    public void set() {
       // if (traceContext != null) {
            //TraceContextHolder.setContext(traceContext);
       // }
    }

    public void clear() {
        //TraceContextHolder.clearContext();
    }

PreservableContexts と CustomHystrixConcurrencyStrategy のログは出力されません。HelloServer の起動時に startegy を登録しています。

HystrixConcurrencyStrategy strategy = new CustomHystrixConcurrencyStrategy();
        HystrixPlugins.getInstance().registerConcurrencyStrategy(strategy);
        context = HystrixRequestContext.initializeContext();

編集2

Observables のセットアップ方法を更新しました。

    ManagedChannel worldChannel = getChannel("localhost:8081", "helloService-world-client");
    //Async stub instead of blockingStub
    WorldServiceGrpc.WorldServiceStub worldServiceStub = WorldServiceGrpc.newStub(worldChannel);
    WorldCommand worldCommand = new WorldCommand(greeterRequest, worldServiceStub);

    //Call TeamService
    ManagedChannel teamChannel = getChannel("localhost:8082", "helloService-team-client");
    TeamServiceGrpc.TeamServiceStub teamServiceStub = TeamServiceGrpc.newStub(teamChannel);
    //TeamServiceGrpc.TeamServiceBlockingStub teamServiceStub = TeamServiceGrpc.newBlockingStub(teamChannel);
    TeamCommand teamCommand = new TeamCommand(greeterRequest, teamServiceStub);

    try {
        rx.Observable<Greeter.GreeterReply> worldReplyObservable = worldCommand.observe().subscribeOn(Schedulers.computation());
        rx.Observable<Greeter.GreeterReply> teamReplyObservable = teamCommand.observe().subscribeOn(Schedulers.computation());
        Observable.zip(worldReplyObservable, teamReplyObservable, new Func2<Greeter.GreeterReply, Greeter.GreeterReply, Object>() {
            @Override
            public Object call(Greeter.GreeterReply worldReply, Greeter.GreeterReply teamReply) {
                String messageFromWorldService = worldReply.getMessage();
                String idFromWorldService = worldReply.getId();
                logger.info("Response from WorldService  -- {}, id = {}", messageFromWorldService, idFromWorldService);

                String messageFromTeamService = teamReply.getMessage();
                String idFromTeamService = teamReply.getId();
                logger.info("Response from TeamService  -- {}, id = {}", messageFromTeamService, idFromTeamService);

                assert(idFromWorldService.equals(idFromTeamService));
                Greeter.GreeterReply greeterReply = Greeter.GreeterReply.newBuilder().setMessage("Hello" + messageFromWorldService + " from " + messageFromTeamService).setId(idFromWorldService).build();
                logger.info("Final response=" + greeterReply.getMessage());
                responseObserver.onNext(greeterReply);
                responseObserver.onCompleted();
                return null;
            }
        });
    } catch (StatusRuntimeException e) {
        logger.warn("Exception when calling WorldService and/or TeamService\n" +  e);
    }

私は今、奇妙な問題を抱えています.このコードでは実行されないため、TeamCommand と WorldCommand の呼び出しは完了しません:

Observable.zip(worldReplyObservable, teamReplyObservable, new Func2<Greeter.GreeterReply, Greeter.GreeterReply, Object>() {
                @Override
                public Object call(Greeter.GreeterReply worldReply, Greeter.GreeterReply teamReply) {
                    String messageFromWorldService = worldReply.getMessage();

また、フォールバックがある場合、hystrix-timer スレッドには MDC がありません。

rx-java - Hystrix Observables で traceIds を渡す方法は?

2 に答える 2

Related

Reference