使用google云跟踪API

Question

我试图使用grpc客户端的StackDrive/谷歌云跟踪定义在https://github.com/googleapis/googleapis/blob/master/google/devtools/cloudtrace/v1/trace.proto

我正在发送协议

traces {
  traces {
    project_id: "brightcove-rna-master"
    trace_id: "A096D4956A424EEB98AE7863505B1E1F"
    spans {
      span_id: 1
      kind: RPC_CLIENT
      name: "/bigtableAapiGrpcstaging_aggregated/queryDataSetInternal/buildScan"
      start_time {
        seconds: 1459230665
        nanos: 705000000
      }
      end_time {
        seconds: 1459230666
        nanos: 416000000
      }
      labels {
        key: "videocloud/account"
        value: "4523394965001"
      }
      labels {
        key: "videocloud/protocol"
        value: "2"
      }
      labels {
        key: "videocloud/dimensions"
        value: "protocol,account"
      }
    }
    spans {
      span_id: 2
      kind: RPC_SERVER
      name: "/bigtableAapiGrpcstaging_aggregated/queryDataSetInternal/aggregateScan"
      start_time {
        seconds: 1459230666
        nanos: 420000000
      }
      end_time {
        seconds: 1459230667
        nanos: 753000000
      }
      labels {
        key: "videocloud/account"
        value: "4523394965001"
      }
      labels {
        key: "videocloud/protocol"
        value: "2"
      }
      labels {
        key: "videocloud/dimensions"
        value: "protocol,account"
      }
      labels {
        key: "bigtable/rowCount"
        value: "339"
      }
      labels {
        key: "bigtable/responseCount"
        value: "136"
      }
    }
    spans {
      kind: RPC_SERVER
      name: "/bigtableAapiGrpcstaging_aggregated/queryDataSetInternal"
      start_time {
        seconds: 1459230665
        nanos: 556000000
      }
      end_time {
        seconds: 1459230667
        nanos: 754000000
      }
      labels {
        key: "account"
        value: "4523394965001"
      }
    }
  }
}

但我得到的唯一回报是这个例外：

[WARN ] [2016-03-28 22:51:09,330] [grpc-default-executor-0] rna.api.server.ServerImpl    Unable to send trace to google
io.grpc.StatusRuntimeException: CANCELLED
    at io.grpc.Status.asRuntimeException(Status.java:431)
    at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:358)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$3.runInContext(ClientCallImpl.java:462)
    at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:54)
    at io.grpc.internal.SerializingExecutor$TaskRunner.run(SerializingExecutor.java:154)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

同样，我尝试了以下ListTrace请求

traceClient.listTraces(ListTracesRequest.newBuilder()
                .setProjectId(projectId)
                .setView(ListTracesRequest.ViewType.COMPLETE)
                .setStartTime(getEpoch())
                .setEndTime(getCurrentTime())
                .build());

并获得：

java.util.concurrent.ExecutionException: io.grpc.StatusRuntimeException: UNIMPLEMENTED: GRPC target method is not implemented.
    at com.google.common.util.concurrent.AbstractFuture.getDoneValue(AbstractFuture.java:476)
    at com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:455)
    at com.brightcove.rna.api.server.ServerImpl.sendTraceAsync(ServerImpl.java:143)
    at com.brightcove.rna.api.server.ServerImpl.queryDataset(ServerImpl.java:116)
    at com.brightcove.rna.api.AnalyticsAPIGrpc$1.invoke(AnalyticsAPIGrpc.java:152)
    at com.brightcove.rna.api.AnalyticsAPIGrpc$1.invoke(AnalyticsAPIGrpc.java:147)
    at io.grpc.stub.ServerCalls$1$1.onHalfClose(ServerCalls.java:147)
    at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:255)
    at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$2.runInContext(ServerImpl.java:458)
    at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:54)
    at io.grpc.internal.SerializingExecutor$TaskRunner.run(SerializingExecutor.java:154)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: io.grpc.StatusRuntimeException: UNIMPLEMENTED: GRPC target method is not implemented.
    at io.grpc.Status.asRuntimeException(Status.java:431)
    at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:358)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$3.runInContext(ClientCallImpl.java:462)
    ... 5 more

我对API非常陌生，所以我不知道我在这里做错了什么。是否有我遗漏的配置值或类似内容。

更新生成跨距/跟踪的代码。该代码表示一个Grpc服务，该服务获取请求并查询云BigTable表。我的目的是跟踪请求的不同方面：

public class ServerImpl implements AnalyticsAPIGrpc.AnalyticsAPI {
    private static final Logger logger = Logger.getLogger(ServerImpl.class);

    private Connection _connection = null;
    private TraceServiceFutureClient traceClient;
    private String projectId;

    @Override
    public void queryDataset(APIRequest request, StreamObserver<APIResponse> responseObserver) {
        APIResponse.Builder response = APIResponse.newBuilder();
        List<TraceSpan> spans = Lists.newArrayList();
        if (request.getTraceToken() != null) {
            response.setTraceToken(request.getTraceToken());
        }

        try {
            spans = queryDataSetInternal(request, response);
            responseObserver.onNext(response.build());
            responseObserver.onCompleted();
        } catch (Exception ex) {
            responseObserver.onError(ex);
        } finally {
            // send trace
            sendTraceAsync(request.getTraceToken(), spans);
        }
    }

    private ListenableFuture<Empty> sendTraceAsync(String traceId, List<TraceSpan> spans) {
        if (spans == null || spans.isEmpty()) {
            return Futures.immediateFuture(Empty.getDefaultInstance());
        }

        PatchTracesRequest patchTracesRequest = PatchTracesRequest.newBuilder()
                .setProjectId(projectId)
                .setTraces(Traces.newBuilder().addTraces(
                        Trace.newBuilder()
                                .setProjectId(projectId)
                                .setTraceId(traceId.replaceAll("-", "").toUpperCase())
                                .addAllSpans(spans)))
                .build();

        if (logger.isTraceEnabled()) {
            logger.trace("Sending trace: " + patchTracesRequest.toString());
        }

        ListenableFuture<Empty> future = traceClient.patchTraces(patchTracesRequest);
        // add callback for logging result
        Futures.addCallback(future, new FutureCallback<Empty>() {
            @Override
            public void onSuccess(@Nullable Empty result) {
                logger.trace("Trace successfully sent to google");
            }

            @Override
            public void onFailure(Throwable t) {
                logger.warn("Unable to send trace to google", t);
            }
        });
        return future;
    }

    private Connection getConnection() throws IOException {
        return this._connection;
    }

    private Scan createScan(APIRequest request, String resourceName) {
        return ScanBuilder.of(
                request.getAccount(),
                resourceName,
                request.getStartTime(), request.getEndTime())
                .build();
    }

    private List<TraceSpan> queryDataSetInternal(APIRequest request, APIResponse.Builder response) throws IOException {
        AtomicLong spanIdCounter = new AtomicLong(0L);
        String parentTraceName = "/api-qa/queryDataSetInternal";
        TraceSpan.Builder parentSpan =
                TraceSpan.newBuilder()
                        .setSpanId(spanIdCounter.getAndIncrement())
                        .setStartTime(getCurrentTime())
                        .setKind(TraceSpan.SpanKind.RPC_SERVER)
                        .setName(parentTraceName)
                        .putAllLabels(ImmutableMap.of("account", request.getAccount()));

        Connection connection = this.getConnection();
        List<TraceSpan> traceSpanList = Lists.newArrayList();
        try (Table table = connection.getTable("tableName")) {
            /// create scan ///
            TraceSpan.Builder traceSpan = TraceSpan.newBuilder()
                    .setSpanId(spanIdCounter.getAndIncrement())
                    .setKind(TraceSpan.SpanKind.RPC_CLIENT)
                    .setName(parentTraceName + "/buildScan")
                    .setParentSpanId(parentSpan.getParentSpanId())
                    .setStartTime(getCurrentTime());

            AtomicInteger count = new AtomicInteger(0);
            // add trace span
            String dimensionStr = Joiner.on(',').join(request.getDimensionsList());
            traceSpan.putAllLabels(ImmutableMap.of(
                    "videocloud/account",    request.getAccount(),
                    "videocloud/protocol",   request.getProtocol(),
                    "videocloud/dimensions", dimensionStr));

            // scan the response and send a stream of rows back
            Scan scan = createScan(request, getResourceName(request));
            logger.debug("Using bigtable scan: " + scan.toJSON());
            ResultScanner scanner = table.getScanner(scan);

            // record trace
            traceSpanList.add(traceSpan.setEndTime(getCurrentTime()).build());

            /// perform aggregation ///
            Timestamp startTS = getCurrentTime();
            List<Result> results = StreamSupport.stream(scanner.spliterator(), false)
                                                .collect(Collectors.toList());
            response.addAllRows(results);

            // record trace
            traceSpan = TraceSpan.newBuilder()
                    .setSpanId(spanIdCounter.getAndIncrement())
                    .setKind(TraceSpan.SpanKind.RPC_SERVER)
                    .setName(parentTraceName + "/aggregateScan")
                    .setParentSpanId(parentSpan.getParentSpanId())
                    .setStartTime(startTS)
                    .setEndTime(getCurrentTime())
                    .putAllLabels(ImmutableMap.of(
                            "videocloud/account",     request.getAccount(),
                            "videocloud/protocol",    request.getProtocol(),
                            "videocloud/dimensions",  dimensionStr,
                            "bigtable/rowCount",      String.valueOf(count.get()),
                            "bigtable/responseCount", String.valueOf(response.getRowsCount())));
            traceSpanList.add(traceSpan.build());

            response.setStatus(APIResponse.Status.OK)
                    .setDetails(String.format("Processed %d results from BigTable", count.get()));

        } finally {
            parentSpan.setEndTime(getCurrentTime());
            traceSpanList.add(parentSpan.build());
        }
        return traceSpanList;
    }
}

Answer 1

Ankur，我在云bigtable github中添加了一个问题。