support logs in the fake test backend (#3374)

* support logs in the fake test backend

* fix copypasta

smoke-tests/fake-backend/src/main/java/io/opentelemetry/smoketest/fakebackend/FakeBackendMain.java

@@ -30,6 +30,7 @@ import com.linecorp.armeria.server.Server;
 import com.linecorp.armeria.server.grpc.GrpcService;
 import com.linecorp.armeria.server.healthcheck.HealthCheckService;
 import io.netty.buffer.ByteBufOutputStream;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
 import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;
 import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
 import java.io.IOException;
@@ -68,6 +69,15 @@ public class FakeBackendMain {
             marshaller.writeValue(value, gen);
           }
         });
+    serializers.addSerializer(
+        new StdSerializer<>(ExportLogsServiceRequest.class) {
+          @Override
+          public void serialize(
+              ExportLogsServiceRequest value, JsonGenerator gen, SerializerProvider provider)
+              throws IOException {
+            marshaller.writeValue(value, gen);
+          }
+        });
     module.setSerializers(serializers);
     mapper.addModule(module);
     OBJECT_MAPPER = mapper.build();
@@ -76,18 +86,21 @@ public class FakeBackendMain {
   public static void main(String[] args) {
     var traceCollector = new FakeTraceCollectorService();
     var metricsCollector = new FakeMetricsCollectorService();
+    var logsCollector = new FakeLogsCollectorService();
     var server =
         Server.builder()
             .http(8080)
             .service(GrpcService.builder()
                 .addService(traceCollector)
                 .addService(metricsCollector)
+                .addService(logsCollector)
                 .build())
             .service(
                 "/clear",
                 (ctx, req) -> {
                   traceCollector.clearRequests();
                   metricsCollector.clearRequests();
+                  logsCollector.clearRequests();
                   return HttpResponse.of(HttpStatus.OK);
                 })
             .service(
@@ -108,6 +121,15 @@ public class FakeBackendMain {
                   return HttpResponse.of(
                       HttpStatus.OK, MediaType.JSON, HttpData.wrap(buf.buffer()));
                 })
+            .service(
+                "/get-logs",
+                (ctx, req) -> {
+                  var requests = logsCollector.getRequests();
+                  var buf = new ByteBufOutputStream(ctx.alloc().buffer());
+                  OBJECT_MAPPER.writeValue((OutputStream) buf, requests);
+                  return HttpResponse.of(
+                      HttpStatus.OK, MediaType.JSON, HttpData.wrap(buf.buffer()));
+                })
             .service("/health", HealthCheckService.of())
             .build();
 

smoke-tests/fake-backend/src/main/java/io/opentelemetry/smoketest/fakebackend/FakeLogsCollectorService.java

@@ -0,0 +1,31 @@
+package io.opentelemetry.smoketest.fakebackend;
+
+import com.google.common.collect.ImmutableList;
+import io.grpc.stub.StreamObserver;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceResponse;
+import io.opentelemetry.proto.collector.logs.v1.LogsServiceGrpc;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingDeque;
+
+public class FakeLogsCollectorService extends LogsServiceGrpc.LogsServiceImplBase{
+
+  private BlockingQueue<ExportLogsServiceRequest> exportRequests = new LinkedBlockingDeque<>();
+
+  List<ExportLogsServiceRequest> getRequests() {
+    return ImmutableList.copyOf(exportRequests);
+  }
+
+  void clearRequests() {
+    exportRequests.clear();
+  }
+
+  @Override
+  public void export(ExportLogsServiceRequest request,
+      StreamObserver<ExportLogsServiceResponse> responseObserver) {
+    exportRequests.add(request);
+    responseObserver.onNext(ExportLogsServiceResponse.getDefaultInstance());
+    responseObserver.onCompleted();
+  }
+}