diff --git a/instrumentation-api/build.gradle.kts b/instrumentation-api/build.gradle.kts index 0d10f0a89707..b01b458e9c7c 100644 --- a/instrumentation-api/build.gradle.kts +++ b/instrumentation-api/build.gradle.kts @@ -41,4 +41,4 @@ dependencies { testImplementation("org.awaitility:awaitility") testImplementation("io.opentelemetry:opentelemetry-sdk-metrics") testImplementation("io.opentelemetry:opentelemetry-sdk-testing") -} +} \ No newline at end of file diff --git a/instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/filibuster/OpenTelemetryContextStorageConstants.java b/instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/filibuster/OpenTelemetryContextStorageConstants.java new file mode 100644 index 000000000000..d35b25b9f58a --- /dev/null +++ b/instrumentation-api/src/main/java/io/opentelemetry/instrumentation/api/filibuster/OpenTelemetryContextStorageConstants.java @@ -0,0 +1,10 @@ +package io.opentelemetry.instrumentation.api.filibuster; + +import io.opentelemetry.context.ContextKey; + +public class OpenTelemetryContextStorageConstants { + final public static ContextKey VCLOCK_KEY = ContextKey.named("filibuster-vclock"); + final public static ContextKey ORIGIN_VCLOCK_KEY = ContextKey.named("filibuster-origin-vclock"); + final public static ContextKey REQUEST_ID_KEY = ContextKey.named("filibuster-request-id"); + final public static ContextKey EXECUTION_INDEX_KEY = ContextKey.named("filibuster-execution-index"); +} diff --git a/instrumentation/armeria-1.3/library/build.gradle.kts b/instrumentation/armeria-1.3/library/build.gradle.kts index fc599f337d78..dba873fbffec 100644 --- a/instrumentation/armeria-1.3/library/build.gradle.kts +++ b/instrumentation/armeria-1.3/library/build.gradle.kts @@ -6,5 +6,22 @@ plugins { dependencies { library("com.linecorp.armeria:armeria:1.3.0") + library("cloud.filibuster:instrumentation:0.19-SNAPSHOT") + + library("com.github.cliftonlabs:json-simple:2.1.2") + library("org.json:json:20210307") + testImplementation(project(":instrumentation:armeria-1.3:testing")) } + +repositories { + mavenCentral() + + maven { + url = uri("https://maven.pkg.github.com/filibuster-testing/filibuster-java") + credentials { + username = project.findProperty("gpr.user") as String? ?: System.getenv("GITHUB_USERNAME") + password = project.findProperty("gpr.key") as String? ?: System.getenv("GITHUB_TOKEN") + } + } +} diff --git a/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/ArmeriaTracing.java b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/ArmeriaTracing.java index be751aa94ad4..1b4e01e8da3e 100644 --- a/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/ArmeriaTracing.java +++ b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/ArmeriaTracing.java @@ -15,6 +15,7 @@ import java.util.function.Function; /** Entrypoint for tracing Armeria services or clients. */ +@SuppressWarnings("FieldCanBeLocal") public final class ArmeriaTracing { /** Returns a new {@link ArmeriaTracing} configured with the given {@link OpenTelemetry}. */ @@ -41,7 +42,7 @@ public static ArmeriaTracingBuilder newBuilder(OpenTelemetry openTelemetry) { * com.linecorp.armeria.client.ClientBuilder#decorator(Function)}. */ public Function newClientDecorator() { - return client -> new OpenTelemetryClient(client, clientInstrumenter); + return client -> new OpenTelemetryFilibusterDecoratingHttpClient(client, clientInstrumenter); } /** @@ -49,6 +50,6 @@ public static ArmeriaTracingBuilder newBuilder(OpenTelemetry openTelemetry) { * HttpService#decorate(Function)}. */ public Function newServiceDecorator() { - return service -> new OpenTelemetryService(service, serverInstrumenter); + return service -> new OpenTelemetryFilibusterDecoratingHttpService(service, serverInstrumenter); } } diff --git a/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryContextStorage.java b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryContextStorage.java new file mode 100644 index 000000000000..b32fd7b9c44c --- /dev/null +++ b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryContextStorage.java @@ -0,0 +1,93 @@ +package io.opentelemetry.instrumentation.armeria.v1_3; + +import cloud.filibuster.instrumentation.datatypes.VectorClock; +import cloud.filibuster.instrumentation.storage.ContextStorage; +import io.opentelemetry.context.Context; +import javax.annotation.Nullable; +import java.util.logging.Level; +import java.util.logging.Logger; + +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.EXECUTION_INDEX_KEY; +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.ORIGIN_VCLOCK_KEY; +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.REQUEST_ID_KEY; +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.VCLOCK_KEY; + +public class OpenTelemetryContextStorage implements ContextStorage { + private static final Logger logger = Logger.getLogger(OpenTelemetryContextStorage.class.getName()); + + private Context context; + + // Context.current() should maybe be cached, who knows? + + public OpenTelemetryContextStorage() { + this.context = Context.current(); + } + + public Context getContext() { + return this.context; + } + + @Override + @Nullable + public String getRequestId() { + return Context.current().get(REQUEST_ID_KEY); + } + + @Override + @Nullable + public VectorClock getVectorClock() { + String vectorClockStr = Context.current().get(VCLOCK_KEY); + + VectorClock newVclock = new VectorClock(); + + if (vectorClockStr != null) { + newVclock.fromString(vectorClockStr); + } + + return newVclock; + } + + @Override + @Nullable + public VectorClock getOriginVectorClock() { + String originVectorClockStr = Context.current().get(ORIGIN_VCLOCK_KEY); + + VectorClock newVclock = new VectorClock(); + + if (originVectorClockStr != null) { + newVclock.fromString(originVectorClockStr); + } + + return newVclock; + } + + @Override + @Nullable + public String getExecutionIndex() { + return Context.current().get(EXECUTION_INDEX_KEY); + } + + @Override + public void setRequestId(String requestId) { + this.context = this.context.with(REQUEST_ID_KEY, requestId); + logger.log(Level.SEVERE, "setRequestId: " + requestId); + } + + @Override + public void setVectorClock(VectorClock vectorClock) { + this.context = this.context.with(VCLOCK_KEY, vectorClock.toString()); + logger.log(Level.SEVERE, "setVectorClock: " + vectorClock); + } + + @Override + public void setOriginVectorClock(VectorClock originVectorClock) { + this.context = this.context.with(ORIGIN_VCLOCK_KEY, originVectorClock.toString()); + logger.log(Level.SEVERE, "setOriginVectorClock: " + originVectorClock); + } + + @Override + public void setExecutionIndex(String executionIndex) { + this.context = this.context.with(EXECUTION_INDEX_KEY, executionIndex); + logger.log(Level.SEVERE, "setExecutionIndex: " + executionIndex); + } +} diff --git a/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryFilibusterDecoratingHttpClient.java b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryFilibusterDecoratingHttpClient.java new file mode 100644 index 000000000000..24c7d5ccec58 --- /dev/null +++ b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryFilibusterDecoratingHttpClient.java @@ -0,0 +1,81 @@ +package io.opentelemetry.instrumentation.armeria.v1_3; + +import cloud.filibuster.instrumentation.libraries.armeria.http.FilibusterDecoratingHttpClient; +import com.linecorp.armeria.client.ClientRequestContext; +import com.linecorp.armeria.client.HttpClient; +import com.linecorp.armeria.common.HttpRequest; +import com.linecorp.armeria.common.HttpResponse; +import com.linecorp.armeria.common.logging.RequestLog; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter; + +import javax.annotation.Nullable; +import java.util.logging.Level; +import java.util.logging.Logger; + +public class OpenTelemetryFilibusterDecoratingHttpClient extends FilibusterDecoratingHttpClient { + private static final Logger logger = Logger.getLogger(OpenTelemetryFilibusterDecoratingHttpClient.class.getName()); + + @Nullable + private final Instrumenter clientInstrumentor; + + @SuppressWarnings("NullAway") + private Context parentContext; + + @SuppressWarnings("NullAway") + private Context context; + + public OpenTelemetryFilibusterDecoratingHttpClient(HttpClient delegate, String serviceName, Instrumenter clientInstrumentor) { + super(delegate); + this.serviceName = serviceName; + this.clientInstrumentor = clientInstrumentor; + this.contextStorage = new OpenTelemetryContextStorage(); + } + + public OpenTelemetryFilibusterDecoratingHttpClient(HttpClient delegate, Instrumenter clientInstrumentor) { + super(delegate); + this.serviceName = System.getenv("SERVICE_NAME"); + this.clientInstrumentor = clientInstrumentor; + this.contextStorage = new OpenTelemetryContextStorage(); + } + + @Override + protected void setupContext(ClientRequestContext ctx, HttpRequest req) { + this.parentContext = Context.current(); + this.context = Context.current(); + + logger.log(Level.INFO, "****************************************************************"); + logger.log(Level.SEVERE, "CLIENT parentContext: " + parentContext.toString()); + logger.log(Level.INFO, "****************************************************************"); + + if (clientInstrumentor != null) { + this.context = clientInstrumentor.start(Context.current(), ctx); + } + + logger.log(Level.INFO, "****************************************************************"); + logger.log(Level.SEVERE, "CLIENT context: " + context.toString()); + logger.log(Level.INFO, "****************************************************************"); + } + + @Override + protected void contextWhenComplete(ClientRequestContext ctx) { + ctx.log().whenComplete().thenAccept(log -> { + if (clientInstrumentor != null) { + clientInstrumentor.end(context, ctx, log, log.responseCause()); + } + }); + } + + @Override + protected HttpResponse delegateWithContext(ClientRequestContext ctx, HttpRequest req) throws Exception { + HttpResponse response; + + try (Scope ignored = context.makeCurrent()) { + logger.log(Level.INFO, "!!!!!!! with context: " + context.toString()); + response = unwrap().execute(ctx, req); + } + + return response; + } +} diff --git a/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryFilibusterDecoratingHttpService.java b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryFilibusterDecoratingHttpService.java new file mode 100644 index 000000000000..d7d152550a78 --- /dev/null +++ b/instrumentation/armeria-1.3/library/src/main/java/io/opentelemetry/instrumentation/armeria/v1_3/OpenTelemetryFilibusterDecoratingHttpService.java @@ -0,0 +1,65 @@ +package io.opentelemetry.instrumentation.armeria.v1_3; + +import cloud.filibuster.instrumentation.libraries.armeria.http.FilibusterDecoratingHttpService; +import com.linecorp.armeria.common.HttpRequest; +import com.linecorp.armeria.common.HttpResponse; +import com.linecorp.armeria.common.logging.RequestLog; +import com.linecorp.armeria.server.HttpService; +import com.linecorp.armeria.server.ServiceRequestContext; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter; + +import javax.annotation.Nullable; +import java.util.logging.Logger; + +public class OpenTelemetryFilibusterDecoratingHttpService extends FilibusterDecoratingHttpService { + private static final Logger logger = Logger.getLogger(OpenTelemetryFilibusterDecoratingHttpService.class.getName()); + + @Nullable + private final Instrumenter serverInstrumenter; + + @SuppressWarnings("NullAway") + private Context context; + + public OpenTelemetryFilibusterDecoratingHttpService(HttpService delegate, String serviceName, Instrumenter serverInstrumenter) { + super(delegate); + this.serviceName = serviceName; + this.serverInstrumenter = serverInstrumenter; + this.contextStorage = new OpenTelemetryContextStorage(); + } + + public OpenTelemetryFilibusterDecoratingHttpService(HttpService delegate, Instrumenter serverInstrumenter) { + super(delegate); + this.serviceName = System.getenv("SERVICE_NAME"); + this.serverInstrumenter = serverInstrumenter; + this.contextStorage = new OpenTelemetryContextStorage(); + } + + @Override + protected void setupContext(ServiceRequestContext ctx, HttpRequest req) { + OpenTelemetryContextStorage openTelemetryContextStorage = (OpenTelemetryContextStorage) this.contextStorage; + context = openTelemetryContextStorage.getContext(); + + if (serverInstrumenter != null) { + context = serverInstrumenter.start(context, ctx); + } + } + + @Override + protected void contextWhenComplete(ServiceRequestContext ctx) { + ctx.log().whenComplete().thenAccept(log -> { + if (serverInstrumenter != null) { + serverInstrumenter.end(context, ctx, log, log.responseCause()); + } + }); + } + + @Override + protected HttpResponse delegateWithContext(ServiceRequestContext ctx, HttpRequest req) throws Exception { + try (Scope ignored = context.makeCurrent()) { + HttpService delegate = (HttpService) unwrap(); + return delegate.serve(ctx, req); + } + } +} diff --git a/instrumentation/grpc-1.6/library/build.gradle.kts b/instrumentation/grpc-1.6/library/build.gradle.kts index e67788d53241..d94d0ff18eee 100644 --- a/instrumentation/grpc-1.6/library/build.gradle.kts +++ b/instrumentation/grpc-1.6/library/build.gradle.kts @@ -7,11 +7,28 @@ val grpcVersion = "1.6.0" dependencies { library("io.grpc:grpc-core:$grpcVersion") + library("cloud.filibuster:instrumentation:0.19-SNAPSHOT") + testLibrary("io.grpc:grpc-netty:$grpcVersion") testLibrary("io.grpc:grpc-protobuf:$grpcVersion") testLibrary("io.grpc:grpc-services:$grpcVersion") testLibrary("io.grpc:grpc-stub:$grpcVersion") + library("com.github.cliftonlabs:json-simple:2.1.2") + library("org.json:json:20210307") + testImplementation("org.assertj:assertj-core") testImplementation(project(":instrumentation:grpc-1.6:testing")) } + +repositories { + mavenCentral() + + maven { + url = uri("https://maven.pkg.github.com/filibuster-testing/filibuster-java") + credentials { + username = project.findProperty("gpr.user") as String? ?: System.getenv("GITHUB_USERNAME") + password = project.findProperty("gpr.key") as String? ?: System.getenv("GITHUB_TOKEN") + } + } +} diff --git a/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTracing.java b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTracing.java index 2ed8cf35b490..8d6197697955 100644 --- a/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTracing.java +++ b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/GrpcTracing.java @@ -13,6 +13,7 @@ import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter; /** Entrypoint for tracing gRPC servers or clients. */ +@SuppressWarnings("FieldCanBeLocal") public final class GrpcTracing { /** Returns a new {@link GrpcTracing} configured with the given {@link OpenTelemetry}. */ @@ -46,7 +47,7 @@ public static GrpcTracingBuilder newBuilder(OpenTelemetry openTelemetry) { * io.grpc.ManagedChannelBuilder#intercept(ClientInterceptor...)}. */ public ClientInterceptor newClientInterceptor() { - return new TracingClientInterceptor(clientInstrumenter, propagators); + return new OpenTelemetryFilibusterClientInterceptor(clientInstrumenter, propagators); } /** @@ -54,6 +55,6 @@ public ClientInterceptor newClientInterceptor() { * io.grpc.ServerBuilder#intercept(ServerInterceptor)}. */ public ServerInterceptor newServerInterceptor() { - return new TracingServerInterceptor(serverInstrumenter, captureExperimentalSpanAttributes); + return new OpenTelemetryFilibusterServerInterceptor(serverInstrumenter); } } diff --git a/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryContextStorage.java b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryContextStorage.java new file mode 100644 index 000000000000..41b3ebb6c5c0 --- /dev/null +++ b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryContextStorage.java @@ -0,0 +1,93 @@ +package io.opentelemetry.instrumentation.grpc.v1_6; + +import cloud.filibuster.instrumentation.datatypes.VectorClock; +import cloud.filibuster.instrumentation.storage.ContextStorage; +import io.opentelemetry.context.Context; +import javax.annotation.Nullable; +import java.util.logging.Level; +import java.util.logging.Logger; + +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.EXECUTION_INDEX_KEY; +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.ORIGIN_VCLOCK_KEY; +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.REQUEST_ID_KEY; +import static io.opentelemetry.instrumentation.api.filibuster.OpenTelemetryContextStorageConstants.VCLOCK_KEY; + +public class OpenTelemetryContextStorage implements ContextStorage { + private static final Logger logger = Logger.getLogger(OpenTelemetryContextStorage.class.getName()); + + private Context context; + + // Context.current() should maybe be cached, who knows? + + public OpenTelemetryContextStorage() { + this.context = Context.current(); + } + + public Context getContext() { + return this.context; + } + + @Override + @Nullable + public String getRequestId() { + return Context.current().get(REQUEST_ID_KEY); + } + + @Override + @Nullable + public VectorClock getVectorClock() { + String vectorClockStr = Context.current().get(VCLOCK_KEY); + + VectorClock newVclock = new VectorClock(); + + if (vectorClockStr != null) { + newVclock.fromString(vectorClockStr); + } + + return newVclock; + } + + @Override + @Nullable + public VectorClock getOriginVectorClock() { + String originVectorClockStr = Context.current().get(ORIGIN_VCLOCK_KEY); + + VectorClock newVclock = new VectorClock(); + + if (originVectorClockStr != null) { + newVclock.fromString(originVectorClockStr); + } + + return newVclock; + } + + @Override + @Nullable + public String getExecutionIndex() { + return Context.current().get(EXECUTION_INDEX_KEY); + } + + @Override + public void setRequestId(String requestId) { + this.context = this.context.with(REQUEST_ID_KEY, requestId); + logger.log(Level.SEVERE, "setRequestId: " + requestId); + } + + @Override + public void setVectorClock(VectorClock vectorClock) { + this.context = this.context.with(VCLOCK_KEY, vectorClock.toString()); + logger.log(Level.SEVERE, "setVectorClock: " + vectorClock); + } + + @Override + public void setOriginVectorClock(VectorClock originVectorClock) { + this.context = this.context.with(ORIGIN_VCLOCK_KEY, originVectorClock.toString()); + logger.log(Level.SEVERE, "setOriginVectorClock: " + originVectorClock); + } + + @Override + public void setExecutionIndex(String executionIndex) { + this.context = this.context.with(EXECUTION_INDEX_KEY, executionIndex); + logger.log(Level.SEVERE, "setExecutionIndex: " + executionIndex); + } +} diff --git a/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryFilibusterClientInterceptor.java b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryFilibusterClientInterceptor.java new file mode 100644 index 000000000000..4dd616907c19 --- /dev/null +++ b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryFilibusterClientInterceptor.java @@ -0,0 +1,458 @@ +package io.opentelemetry.instrumentation.grpc.v1_6; + +import cloud.filibuster.instrumentation.datatypes.Callsite; +import cloud.filibuster.instrumentation.instrumentors.FilibusterClientInstrumentor; +import cloud.filibuster.instrumentation.libraries.grpc.NoopClientCall; +import cloud.filibuster.instrumentation.storage.ContextStorage; +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ClientCall; +import io.grpc.ClientInterceptor; +import io.grpc.ForwardingClientCall; +import io.grpc.ForwardingClientCallListener; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.context.propagation.ContextPropagators; +import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter; +import org.json.JSONObject; + +import javax.annotation.Nullable; +import java.util.HashMap; +import java.util.logging.Level; +import java.util.logging.Logger; + +import static cloud.filibuster.instrumentation.Helper.getDisableInstrumentationFromEnvironment; +import static cloud.filibuster.instrumentation.Helper.getDisableServerCommunicationFromEnvironment; + +public class OpenTelemetryFilibusterClientInterceptor implements ClientInterceptor { + private static final Logger logger = Logger.getLogger(OpenTelemetryFilibusterClientInterceptor.class.getName()); + + @SuppressWarnings("FieldCanBeFinal") + protected String serviceName; + + @SuppressWarnings("FieldCanBeFinal") + protected ContextStorage contextStorage; + + public static Boolean disableServerCommunication = false; + public static Boolean disableInstrumentation = false; + + @Nullable + private final Instrumenter clientInstrumentor; + + @SuppressWarnings("FieldCanBeLocal") + private final ContextPropagators propagators; + + private static boolean shouldInstrument() { + if (disableInstrumentation) { + return false; + } + + return !getDisableInstrumentationFromEnvironment(); + } + + private static boolean shouldCommunicateWithServer() { + if (disableServerCommunication) { + return false; + } + + return !getDisableServerCommunicationFromEnvironment(); + } + + private static Status generateCorrectStatusForAbort(FilibusterClientInstrumentor filibusterClientInstrumentor) { + JSONObject forcedException = filibusterClientInstrumentor.getForcedException(); + JSONObject forcedExceptionMetadata = forcedException.getJSONObject("metadata"); + String codeStr = forcedExceptionMetadata.getString("code"); + Status.Code code = Status.Code.valueOf(codeStr); + Status status = Status.fromCode(code); + return status; + } + + private static Status generateException(FilibusterClientInstrumentor filibusterClientInstrumentor) { + JSONObject forcedException = filibusterClientInstrumentor.getForcedException(); + + // Create the exception to throw. + String exceptionNameString = forcedException.getString("name"); + JSONObject forcedExceptionMetadata = forcedException.getJSONObject("metadata"); + String causeString = forcedExceptionMetadata.getString("cause"); + String codeStr = forcedExceptionMetadata.getString("code"); + Status.Code code = Status.Code.valueOf(codeStr); + + // Notify Filibuster of failure. + HashMap additionalMetadata = new HashMap<>(); + additionalMetadata.put("code", codeStr); + filibusterClientInstrumentor.afterInvocationWithException(exceptionNameString, causeString, additionalMetadata); + + // Return status. + return Status.fromCode(code); + } + + private static Status generateExceptionFromFailureMetadata(FilibusterClientInstrumentor filibusterClientInstrumentor) { + JSONObject failureMetadata = filibusterClientInstrumentor.getFailureMetadata(); + JSONObject exception = failureMetadata.getJSONObject("exception"); + JSONObject exceptionMetadata = exception.getJSONObject("metadata"); + + // Create the exception to throw. + String exceptionNameString = "io.grpc.StatusRuntimeException"; + String codeStr = exceptionMetadata.getString("code"); + Status.Code code = Status.Code.valueOf(codeStr); + StatusRuntimeException status = Status.fromCode(code).asRuntimeException(); + String causeString = ""; + + // Notify Filibuster of failure. + HashMap additionalMetadata = new HashMap<>(); + additionalMetadata.put("name", exceptionNameString); + additionalMetadata.put("code", codeStr); + filibusterClientInstrumentor.afterInvocationWithException(exceptionNameString, causeString, additionalMetadata); + + // Return status. + return Status.fromCode(code); + } + + public OpenTelemetryFilibusterClientInterceptor(Instrumenter clientInstrumentor, ContextPropagators propagators) { + this.serviceName = System.getenv("SERVICE_NAME"); + this.contextStorage = new OpenTelemetryContextStorage(); + this.clientInstrumentor = clientInstrumentor; + this.propagators = propagators; + } + + public OpenTelemetryFilibusterClientInterceptor(String serviceName, Instrumenter clientInstrumentor, ContextPropagators propagators) { + this.serviceName = serviceName; + this.contextStorage = new OpenTelemetryContextStorage(); + this.clientInstrumentor = clientInstrumentor; + this.propagators = propagators; + } + + @Override + public ClientCall interceptCall( + MethodDescriptor method, CallOptions callOptions, Channel next) { + + if (method.getType() != MethodDescriptor.MethodType.UNARY) { + return next.newCall(method, callOptions); + } + + GrpcRequest request = new GrpcRequest(method, null, null); + Context parentContext = Context.current(); + + Context context; + + if (clientInstrumentor != null) { + context = clientInstrumentor.start(parentContext, request); + } else { + context = Context.current(); + } + + // return the filibuster client interceptor. + return new ForwardingClientCall() { + @Nullable + private ClientCall delegate; + private Listener responseListener; + + @Nullable + private Metadata headers; + private int requestTokens; + private FilibusterClientInstrumentor filibusterClientInstrumentor; + + @Override + protected ClientCall delegate() { + if (delegate == null) { + throw new UnsupportedOperationException(); + } + return delegate; + } + + @Override + public void start(Listener responseListener, Metadata headers) { + logger.log(Level.INFO, "INSIDE: start!"); + + this.headers = headers; + this.responseListener = responseListener; + } + + @Override + public void request(int requests) { + if (delegate == null) { + requestTokens += requests; + return; + } + super.request(requests); + } + + // This method is invoked with the message from the Client to the Service. + // message: type of the message issued from the Client (e.g., Hello$HelloRequest) + @Override + public void sendMessage(REQUEST message) { + logger.log(Level.INFO, "INSIDE: sendMessage!"); + logger.log(Level.INFO, "message: " + message.toString()); + + try (Scope ignored = context.makeCurrent()) { + // ****************************************************************************************** + // Figure out if we are inside of instrumentation. + // ****************************************************************************************** + + String instrumentationRequestStr = headers.get( + Metadata.Key.of("x-filibuster-instrumentation", Metadata.ASCII_STRING_MARSHALLER)); + logger.log(Level.INFO, "instrumentationRequestStr: " + instrumentationRequestStr); + boolean instrumentationRequest = Boolean.parseBoolean(instrumentationRequestStr); + logger.log(Level.INFO, "instrumentationRequest: " + instrumentationRequest); + + if (! shouldInstrument() || instrumentationRequest) { + delegate = next.newCall(method, callOptions); + super.start(responseListener, headers); + headers = null; + if (requestTokens > 0) { + super.request(requestTokens); + requestTokens = 0; + } + } else { + // ****************************************************************************************** + // Extract callsite information. + // ****************************************************************************************** + + String grpcFullMethodName = method.getFullMethodName(); + String grpcServiceName = grpcFullMethodName.substring(0, grpcFullMethodName.indexOf("/")); + String grpcRpcName = grpcFullMethodName.replace(grpcServiceName + "/", ""); + +// logger.log(Level.INFO, "method: " + method); + logger.log(Level.INFO, "grpcFullMethodName: " + grpcFullMethodName); + logger.log(Level.INFO, "grpcServiceName: " + grpcServiceName); + logger.log(Level.INFO, "grpcRpcName: " + grpcRpcName); + + // ****************************************************************************************** + // Construct preliminary call site information. + // ****************************************************************************************** + + Callsite callsite = new Callsite( + serviceName, + grpcServiceName, + grpcFullMethodName, + message.toString() + ); + + // ****************************************************************************************** + // Prepare for invocation. + // ****************************************************************************************** + + this.filibusterClientInstrumentor = new FilibusterClientInstrumentor( + serviceName, + shouldCommunicateWithServer(), + contextStorage, + callsite + ); + filibusterClientInstrumentor.prepareForInvocation(); + + // ****************************************************************************************** + // Record invocation. + // ****************************************************************************************** + + filibusterClientInstrumentor.beforeInvocation(); + + // ****************************************************************************************** + // Attach metadata to outgoing request. + // ****************************************************************************************** + + logger.log(Level.INFO, "requestId: " + filibusterClientInstrumentor.getOutgoingRequestId()); + + if (filibusterClientInstrumentor.getOutgoingRequestId() != null) { + headers.put( + Metadata.Key.of("x-filibuster-request-id", Metadata.ASCII_STRING_MARSHALLER), + filibusterClientInstrumentor.getOutgoingRequestId() + ); + } + + if (filibusterClientInstrumentor.getGeneratedId() > -1) { + headers.put( + Metadata.Key.of("x-filibuster-generated-id", Metadata.ASCII_STRING_MARSHALLER), + String.valueOf(filibusterClientInstrumentor.getGeneratedId()) + ); + } + + headers.put( + Metadata.Key.of("x-filibuster-vclock", Metadata.ASCII_STRING_MARSHALLER), + filibusterClientInstrumentor.getVectorClock().toString() + ); + headers.put( + Metadata.Key.of("x-filibuster-origin-vclock", Metadata.ASCII_STRING_MARSHALLER), + filibusterClientInstrumentor.getOriginVectorClock().toString() + ); + headers.put( + Metadata.Key.of("x-filibuster-execution-index", Metadata.ASCII_STRING_MARSHALLER), + filibusterClientInstrumentor.getExecutionIndex().toString() + ); + + // ****************************************************************************************** + // Get failure information. + // ****************************************************************************************** + + JSONObject forcedException = filibusterClientInstrumentor.getForcedException(); + JSONObject failureMetadata = filibusterClientInstrumentor.getFailureMetadata(); + + logger.log(Level.INFO, "forcedException: " + forcedException); + logger.log(Level.INFO, "failureMetadata: " + failureMetadata); + + // ****************************************************************************************** + // Setup additional failure headers, if necessary. + // ****************************************************************************************** + + if (forcedException != null) { + JSONObject forcedExceptionMetadata = forcedException.getJSONObject("metadata"); + + if (forcedExceptionMetadata.has("sleep")) { + int sleepInterval = forcedExceptionMetadata.getInt("sleep"); + headers.put( + Metadata.Key.of("x-filibuster-forced-sleep", Metadata.ASCII_STRING_MARSHALLER), + String.valueOf(sleepInterval) + ); + } else { + headers.put( + Metadata.Key.of("x-filibuster-forced-sleep", Metadata.ASCII_STRING_MARSHALLER), + String.valueOf(0) + ); + } + } + + // ****************************************************************************************** + // If we need to override the response, do it now before proceeding. + // ****************************************************************************************** + + if (failureMetadata != null && filibusterClientInstrumentor.shouldAbort()) { + delegate = new NoopClientCall(); + Status status = generateExceptionFromFailureMetadata(filibusterClientInstrumentor); + responseListener.onClose(status, new Metadata()); + return; + } + + // ****************************************************************************************** + // If we need to throw, this is where we throw. + // ****************************************************************************************** + + if (forcedException != null && filibusterClientInstrumentor.shouldAbort()) { + delegate = new NoopClientCall(); + Status status = generateException(filibusterClientInstrumentor); + responseListener.onClose(status, new Metadata()); + return; + } + + delegate = next.newCall(method, callOptions); + super.start(new FilibusterClientCallListener<>( + responseListener, parentContext, context, request, filibusterClientInstrumentor), headers); + headers = null; + if (requestTokens > 0) { + super.request(requestTokens); + requestTokens = 0; + } + } + + super.sendMessage(message); + } catch (Throwable e) { + if (clientInstrumentor != null) { + clientInstrumentor.end(context, request, null, e); + } + + throw e; + } + } + }; + } + + // ********************************************************************* + // Client caller listener. + + @SuppressWarnings("ClassCanBeStatic") + final class FilibusterClientCallListener + extends ForwardingClientCallListener.SimpleForwardingClientCallListener { + + private final FilibusterClientInstrumentor filibusterClientInstrumentor; + private final Context parentContext; + private final Context context; + private final GrpcRequest request; + + FilibusterClientCallListener(ClientCall.Listener delegate, + Context parentContext, + Context context, + GrpcRequest request, + FilibusterClientInstrumentor filibusterClientInstrumentor) { + super(delegate); + this.filibusterClientInstrumentor = filibusterClientInstrumentor; + this.parentContext = parentContext; + this.context = context; + this.request = request; + } + + // invoked on successful response with the message from the Server to the Client + // message: type of message issued from the Server to the Client (e.g., Hello$HelloReply) + @Override + public void onMessage(RESPONSE message) { + logger.log(Level.INFO, "INSIDE: onMessage!"); + logger.log(Level.INFO, "message: " + message); + + if (! filibusterClientInstrumentor.shouldAbort()) { + // Request completed normally, but we want to throw the exception anyway, generate and throw. + generateException(filibusterClientInstrumentor); + } else { + // Request completed normally. + + // Notify Filibuster of complete invocation with the proper response. + String className = message.getClass().getName(); + HashMap returnValueProperties = new HashMap<>(); + filibusterClientInstrumentor.afterInvocationComplete(className, returnValueProperties); + + // Delegate. + try (Scope ignored = context.makeCurrent()) { + delegate().onMessage(message); + } + } + } + + // invoked on an error: status set to a status message + // Status.code = FAILED_PRECONDITION, description = ..., cause = ... + // trailers metadata headers. + @Override + public void onClose(Status status, Metadata trailers) { + if (clientInstrumentor != null) { + clientInstrumentor.end(context, request, status, status.getCause()); + } + + logger.log(Level.INFO, "INSIDE: onClose!"); + logger.log(Level.INFO, "status: " + status); + logger.log(Level.INFO, "trailers: " + trailers); + + if (! filibusterClientInstrumentor.shouldAbort()) { + Status rewrittenStatus = generateCorrectStatusForAbort(filibusterClientInstrumentor); + + try (Scope ignored = parentContext.makeCurrent()) { + delegate().onClose(rewrittenStatus, trailers); + } + } + + if (! status.isOk()) { + // Request completed -- if it completed with a failure, it will be coming here for + // the first time (didn't call onMessage) and therefore, we need to notify the Filibuster + // server that the call completed with failure. If it completed successfully, we would + // have already notified the Filibuster server in the onMessage callback. + + // Notify Filibuster of error. + HashMap additionalMetadata = new HashMap<>(); + additionalMetadata.put("code", status.getCode().toString()); + String exceptionName = "io.grpc.StatusRuntimeException"; + // exception cause is always null, because it doesn't serialize and pass through even if provided. + filibusterClientInstrumentor.afterInvocationWithException(exceptionName, null, additionalMetadata); + } + + try (Scope ignored = parentContext.makeCurrent()) { + delegate().onClose(status, trailers); + } + } + + @Override + public void onReady() { + logger.log(Level.INFO, "INSIDE: onReady!"); + try (Scope ignored = context.makeCurrent()) { + delegate().onReady(); + } + } + } +} diff --git a/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryFilibusterServerInterceptor.java b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryFilibusterServerInterceptor.java new file mode 100644 index 000000000000..7846b457c8b3 --- /dev/null +++ b/instrumentation/grpc-1.6/library/src/main/java/io/opentelemetry/instrumentation/grpc/v1_6/OpenTelemetryFilibusterServerInterceptor.java @@ -0,0 +1,337 @@ +package io.opentelemetry.instrumentation.grpc.v1_6; + +import cloud.filibuster.instrumentation.Helper; +import cloud.filibuster.instrumentation.instrumentors.FilibusterServerInstrumentor; +import cloud.filibuster.instrumentation.storage.ContextStorage; + +import io.grpc.Contexts; +import io.grpc.ForwardingServerCall; +import io.grpc.ForwardingServerCallListener; +import io.grpc.Grpc; +import io.grpc.Metadata; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.ServerInterceptor; +import io.grpc.Status; +import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; +import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter; + +import javax.annotation.Nullable; +import java.util.logging.Level; +import java.util.logging.Logger; + +import static cloud.filibuster.instrumentation.Helper.getDisableInstrumentationFromEnvironment; +import static cloud.filibuster.instrumentation.Helper.getDisableServerCommunicationFromEnvironment; + +public class OpenTelemetryFilibusterServerInterceptor implements ServerInterceptor { + private static final Logger logger = Logger.getLogger(OpenTelemetryFilibusterServerInterceptor.class.getName()); + + @SuppressWarnings("FieldCanBeFinal") + protected String serviceName; + + @SuppressWarnings("FieldCanBeFinal") + protected ContextStorage contextStorage; + + public static Boolean disableServerCommunication = false; + public static Boolean disableInstrumentation = false; + + @Nullable + private String requestId; + + @Nullable + private final Instrumenter serverInstrumentor; + + private static boolean shouldInstrument() { + if (disableInstrumentation) { + return false; + } + + return !getDisableInstrumentationFromEnvironment(); + } + + private static boolean shouldCommunicateWithServer() { + if (disableServerCommunication) { + return false; + } + + return !getDisableServerCommunicationFromEnvironment(); + } + + public OpenTelemetryFilibusterServerInterceptor(Instrumenter serverInstrumentor) { + this.serviceName = System.getenv("SERVICE_NAME"); + this.contextStorage = new OpenTelemetryContextStorage(); + this.serverInstrumentor = serverInstrumentor; + } + + public OpenTelemetryFilibusterServerInterceptor(String serviceName, Instrumenter serverInstrumentor) { + this.serviceName = serviceName; + this.contextStorage = new OpenTelemetryContextStorage(); + this.serverInstrumentor = serverInstrumentor; + } + + // ****************************************************************************************** + // Accessors for metadata. + // ****************************************************************************************** + + public String getRequestIdFromMetadata(Metadata requestHeaders) { + if (this.requestId == null) { + this.requestId = requestHeaders.get(Metadata.Key.of("x-filibuster-request-id", Metadata.ASCII_STRING_MARSHALLER)); + + if (this.requestId == null) { + this.requestId = Helper.generateNewRequestId().toString(); + } + } + + logger.log(Level.INFO, "requestId: " + this.requestId); + return this.requestId; + } + + public String getGeneratedIdFromMetadata(Metadata requestHeaders) { + String generatedId = requestHeaders.get( + Metadata.Key.of("x-filibuster-generated-id", Metadata.ASCII_STRING_MARSHALLER)); + logger.log(Level.INFO, "generateId: " + generatedId); + return generatedId; + } + + public String getVectorClockFromMetadata(Metadata requestHeaders) { + String vclock = requestHeaders.get( + Metadata.Key.of("x-filibuster-vclock", Metadata.ASCII_STRING_MARSHALLER)); + logger.log(Level.INFO, "vclock: " + vclock); + return vclock; + } + + public String getOriginVectorClockFromMetadata(Metadata requestHeaders) { + String originVclock = requestHeaders.get( + Metadata.Key.of("x-filibuster-origin-vclock", Metadata.ASCII_STRING_MARSHALLER)); + logger.log(Level.INFO, "originVclock: " + originVclock); + return originVclock; + } + + public String getExecutionIndexFromMetadata(Metadata requestHeaders) { + String executionIndex = requestHeaders.get( + Metadata.Key.of("x-filibuster-execution-index", Metadata.ASCII_STRING_MARSHALLER)); + logger.log(Level.INFO, "executionIndex: " + executionIndex); + return executionIndex; + } + + @Override + public ServerCall.Listener interceptCall( + ServerCall call, + Metadata headers, + ServerCallHandler next) { + GrpcRequest request = + new GrpcRequest( + call.getMethodDescriptor(), + headers, + call.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR)); + Context context; + + if (serverInstrumentor != null) { + context = serverInstrumentor.start(Context.current(), request); + } else { + context = Context.current(); + } + + try (Scope ignored = context.makeCurrent()) { + if (shouldInstrument()) { + logger.log(Level.INFO, "Entering server interceptor..."); + + // ****************************************************************************************** + // Setup Filibuster instrumentation. + // ****************************************************************************************** + + logger.log(Level.INFO, "!!! Entering constructor."); + + FilibusterServerInstrumentor filibusterServerInstrumentor = new FilibusterServerInstrumentor( + serviceName, + shouldCommunicateWithServer(), + getRequestIdFromMetadata(headers), + getGeneratedIdFromMetadata(headers), + getVectorClockFromMetadata(headers), + getOriginVectorClockFromMetadata(headers), + getExecutionIndexFromMetadata(headers), + contextStorage + ); + + logger.log(Level.INFO, "!!! Leaving constructor."); + + // ****************************************************************************************** + // Force sleep if necessary. + // ****************************************************************************************** + + String sleepIntervalStr = headers.get( + Metadata.Key.of("x-filibuster-forced-sleep", Metadata.ASCII_STRING_MARSHALLER)); + + if (sleepIntervalStr == null) { + sleepIntervalStr = "0"; + } + + int sleepInterval = Integer.parseInt(sleepIntervalStr); + if (sleepInterval > 0) { + try { + Thread.sleep(sleepInterval * 1000L); + } catch (InterruptedException e) { + // Do nothing. + } + } + + // ****************************************************************************************** + // Notify Filibuster before delegation. + // ****************************************************************************************** + + logger.log(Level.INFO, "!!! Entering beforeInvocation."); + + filibusterServerInstrumentor.beforeInvocation(); + + logger.log(Level.INFO, "!!! Leaving beforeInvocation."); + + // ****************************************************************************************** + // Delegate to underlying service. + // ****************************************************************************************** + + logger.log(Level.INFO, "Leaving server interceptor..."); + } + + return new FilibusterServerCall<>(call, context, request, headers).start(headers, next); + } catch (Throwable e) { + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, null, e); + } + + throw e; + } + } + + final class FilibusterServerCall + extends ForwardingServerCall.SimpleForwardingServerCall { + @SuppressWarnings("FieldCanBeLocal") + final private Metadata requestHeaders; + + @SuppressWarnings("FieldCanBeLocal") + private final Context context; + + @SuppressWarnings("FieldCanBeLocal") + private final GrpcRequest request; + + public FilibusterServerCall(ServerCall delegate, Context context, GrpcRequest request, Metadata requestHeaders) { + super(delegate); + this.requestHeaders = requestHeaders; + this.context = context; + this.request = request; + } + + FilibusterServerCallListener start(Metadata headers, ServerCallHandler next) { + return new FilibusterServerCallListener( + Contexts.interceptCall(io.grpc.Context.current(), this, headers, next), context, request); + } + + // ****************************************************************************************** + // Implementation. + // ****************************************************************************************** + + @Override + public void sendMessage(RESPONSE message) { + try (Scope ignored = context.makeCurrent()) { + super.sendMessage(message); + } + } + + @Override + public void close(Status status, Metadata trailers) { + try { + delegate().close(status, trailers); + } catch (Throwable e) { + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, status, e); + } + throw e; + } + + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, status, status.getCause()); + } + } + + @Override + public void sendHeaders(Metadata responseHeaders) { + if (!shouldInstrument()) { + try (Scope ignored = context.makeCurrent()) { + super.sendHeaders(responseHeaders); + } + } else { + try (Scope ignored = context.makeCurrent()) { + super.sendHeaders(responseHeaders); + } + } + } + + final class FilibusterServerCallListener + extends ForwardingServerCallListener.SimpleForwardingServerCallListener { + private final Context context; + private final GrpcRequest request; + + FilibusterServerCallListener(Listener delegate, Context context, GrpcRequest request) { + super(delegate); + this.context = context; + this.request = request; + } + + @Override + public void onMessage(REQUEST message) { + delegate().onMessage(message); + } + + @Override + public void onHalfClose() { + try { + delegate().onHalfClose(); + } catch (Throwable e) { + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, null, e); + } + throw e; + } + } + + @Override + public void onCancel() { + try { + delegate().onCancel(); + } catch (Throwable e) { + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, null, e); + } + throw e; + } + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, null, null); + } + } + + @Override + public void onComplete() { + try { + delegate().onComplete(); + } catch (Throwable e) { + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, null, e); + } + throw e; + } + } + + @Override + public void onReady() { + try { + delegate().onReady(); + } catch (Throwable e) { + if (serverInstrumentor != null) { + serverInstrumentor.end(context, request, null, e); + } + throw e; + } + } + } + } +}