diff --git a/functions-framework-api/pom.xml b/functions-framework-api/pom.xml index 26b466a8..eafad2d1 100644 --- a/functions-framework-api/pom.xml +++ b/functions-framework-api/pom.xml @@ -38,7 +38,7 @@ dev.openfunction.functions functions-framework-api - 1.2.0-SNAPSHOT + 1.3.0-SNAPSHOT UTF-8 diff --git a/functions-framework-api/src/main/java/dev/openfunction/functions/Context.java b/functions-framework-api/src/main/java/dev/openfunction/functions/Context.java index d3863eb9..72adf9d9 100644 --- a/functions-framework-api/src/main/java/dev/openfunction/functions/Context.java +++ b/functions-framework-api/src/main/java/dev/openfunction/functions/Context.java @@ -121,5 +121,5 @@ public interface Context { */ DaprClient getDaprClient(); - CloudEvent packageAsCloudevent(String payload); + byte[] packageAsCloudevent(String payload); } diff --git a/functions-framework-invoker/pom.xml b/functions-framework-invoker/pom.xml index 35a70084..a2d9bdd0 100644 --- a/functions-framework-invoker/pom.xml +++ b/functions-framework-invoker/pom.xml @@ -10,7 +10,7 @@ dev.openfunction.functions functions-framework-invoker - 1.2.0-SNAPSHOT + 1.3.0-SNAPSHOT 3.8.0 @@ -20,6 +20,7 @@ 11 11 2.4.2 + 8.16.0 @@ -60,7 +61,7 @@ dev.openfunction.functions functions-framework-api - 1.2.0-SNAPSHOT + 1.3.0-SNAPSHOT io.cloudevents @@ -179,6 +180,12 @@ jackson-databind 2.14.2 + + + org.apache.skywalking + apm-toolkit-trace + ${skywalking.version} + diff --git a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/FunctionContext.java b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/FunctionContext.java index 8ef77d22..ebe53868 100644 --- a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/FunctionContext.java +++ b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/FunctionContext.java @@ -128,16 +128,6 @@ public void setPluginsTracing(TracingConfig pluginsTracing) { this.pluginsTracing = pluginsTracing; } - public boolean isTracingEnabled() { - if (tracing != null ) { - return tracing.isEnabled(); - } else if (pluginsTracing != null) { - return pluginsTracing.isEnabled(); - } else { - return false; - } - } - public Map getStates() { return states; } diff --git a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/RuntimeContext.java b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/RuntimeContext.java index d116426c..692ebe11 100644 --- a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/RuntimeContext.java +++ b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/RuntimeContext.java @@ -66,26 +66,41 @@ public RuntimeContext(String context, ClassLoader classLoader) throws Exception loadHooks(classLoader); - if (functionContext.isTracingEnabled() && functionContext.getPluginsTracing().getProvider() != null) { - String provider = functionContext.getPluginsTracing().getProvider().getName(); + TracingConfig tracingConfig = getTracingConfig(); + if (tracingConfig != null && tracingConfig.isEnabled() && tracingConfig.getProvider() != null) { + String provider = tracingConfig.getProvider().getName(); if (!Objects.equals(provider, TracingSkywalking) && !Objects.equals(provider, TracingOpentelemetry)) { throw new IllegalArgumentException("unsupported tracing provider " + provider); } switch (provider) { case TracingSkywalking: - tracingProvider = new SkywalkingProvider(); + tracingProvider = new SkywalkingProvider(tracingConfig, + functionContext.getName(), + System.getenv(RuntimeContext.PodNameEnvName), + System.getenv(RuntimeContext.PodNamespaceEnvName)); + break; case TracingOpentelemetry: - tracingProvider = new OpenTelemetryProvider(functionContext.getPluginsTracing(), + tracingProvider = new OpenTelemetryProvider(tracingConfig, functionContext.getName(), System.getenv(RuntimeContext.PodNameEnvName), System.getenv(RuntimeContext.PodNamespaceEnvName)); + break; } } EventFormatProvider.getInstance().registerFormat(new JsonEventFormat()); } + private TracingConfig getTracingConfig() { + TracingConfig tracingConfig = functionContext.getTracing(); + if (tracingConfig != null) { + return tracingConfig; + } + + return functionContext.getPluginsTracing(); + } + private void loadHooks(ClassLoader classLoader) { String[] preHookNames = functionContext.getPreHooks(); if (ArrayUtils.isEmpty(preHookNames)) { diff --git a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/UserContext.java b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/UserContext.java index b8d14f32..e02c60ea 100644 --- a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/UserContext.java +++ b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/context/UserContext.java @@ -101,8 +101,7 @@ public Error send(String outputName, String data) { // binding must be in CloudEvent format, otherwise pubsub cannot parse the data. byte[] payload = data.getBytes(); if (MiddlewaresCloudEventFormatRequired.contains(output.getComponentType())) { - CloudEvent event = packageAsCloudevent(data); - payload = new JsonEventFormat().serialize(event); + payload = packageAsCloudevent(data); } daprClient.invokeBinding(output.getComponentName(), output.getOperation(), payload).block(); @@ -169,14 +168,15 @@ public DaprClient getDaprClient() { } @Override - public CloudEvent packageAsCloudevent(String payload) { - return new CloudEventBuilder() + public byte[] packageAsCloudevent(String payload) { + CloudEvent event = new CloudEventBuilder() .withId(UUID.randomUUID().toString()) .withType("dapr.invoke") .withSource(URI.create("openfunction/invokeBinding")) .withData(payload.getBytes()) .withDataContentType(JsonEventFormat.CONTENT_TYPE) .build(); + return new JsonEventFormat().serialize(event); } @Override diff --git a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/OpenTelemetryProvider.java b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/OpenTelemetryProvider.java index 50d7a107..781bd398 100644 --- a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/OpenTelemetryProvider.java +++ b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/OpenTelemetryProvider.java @@ -48,7 +48,7 @@ public class OpenTelemetryProvider implements TracingProvider { private static final String Protocol_HTTP = "http"; private final String functionName; - private final Map tags; + private Map tags; private final Map baggage; private final TextMapGetter> getter; @@ -107,6 +107,9 @@ public String get(Map carrier, @NotNull String key) { }; tags = config.getTags(); + if (tags == null) { + tags = new HashMap<>(); + } if (!Objects.equals(pod, "")) { tags.put("instance", pod); } @@ -238,7 +241,7 @@ public void executeWithTracing(UserContext ctx, Callback callback) throws Except executeWithTracing(ctx.getFunctionClass().getSimpleName(), kind, tags, callback); } - public void executeWithTracing(Map carrier, Callback callback) throws Exception { + private void executeWithTracing(Map carrier, Callback callback) throws Exception { TextMapPropagator propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator(); Context parentContext = propagator.extract(Context.root(), carrier, getter); Tracer tracer = GlobalOpenTelemetry.getTracer(OTEL_LIBRARY_NAME, OTEL_LIBRARY_VERSION); diff --git a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/SkywalkingProvider.java b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/SkywalkingProvider.java index 8b102300..d99abd40 100644 --- a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/SkywalkingProvider.java +++ b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/tracing/SkywalkingProvider.java @@ -18,61 +18,164 @@ import dev.openfunction.functions.*; import dev.openfunction.invoker.Callback; +import dev.openfunction.invoker.context.TracingConfig; import dev.openfunction.invoker.context.UserContext; import io.cloudevents.CloudEvent; +import org.apache.skywalking.apm.toolkit.trace.*; -import java.util.Map; +import java.util.*; public class SkywalkingProvider implements TracingProvider { /** * ... */ - private static final int componentIDOpenFunction = 5013; - -// private final Map tags; -// private final Map baggage; - - public SkywalkingProvider() { - +// private static final int componentIDOpenFunction = 5013; + + private final String functionName; + private Map tags; + private final Map baggage; + + public SkywalkingProvider(TracingConfig config, String functionName, String pod, String namespace) { + this.functionName = functionName; + + tags = config.getTags(); + tags = config.getTags(); + if (tags == null) { + tags = new HashMap<>(); + } + if (!Objects.equals(pod, "")) { + tags.put("instance", pod); + } + if (!Objects.equals(namespace, "")) { + tags.put("namespace", pod); + } + baggage = config.getBaggage(); } @Override public void executeWithTracing(HttpRequest httpRequest, Callback callback) throws Exception { - + Map carrier = new HashMap<>(); + Map> headers = httpRequest.getHeaders(); + if (headers != null) { + for (String key : headers.keySet()) { + if (headers.get(key).size() > 0) { + carrier.put(key, headers.get(key).get(0)); + } + } + } + + HashMap newTags = new HashMap<>(tags); + newTags.put("Method", httpRequest.getMethod()); + newTags.put("URI", httpRequest.getUri()); + + executeWithTracing(carrier, newTags, callback); } @Override public void executeWithTracing(CloudEvent event, Callback callback) throws Exception { - + Map carrier = new HashMap<>(); + for (String key : event.getExtensionNames()) { + Object obj = event.getExtension(key); + carrier.put(key, obj == null ? "" : obj.toString()); + } + executeWithTracing(carrier, tags, callback); } @Override public void executeWithTracing(BindingEvent event, Callback callback) throws Exception { - + executeWithTracing(new HashMap<>(), tags, callback); } @Override public void executeWithTracing(TopicEvent event, Callback callback) throws Exception { + executeWithTracing(new HashMap<>(), tags, callback); + } + private void executeWithTracing(Map carrier, Map tags, Callback callback) throws Exception { + ContextCarrierRef contextCarrierRef = new ContextCarrierRef(); + if (carrier != null) { + CarrierItemRef next = contextCarrierRef.items(); + while (next.hasNext()) { + next = next.next(); + if (carrier.get(next.getHeadKey()) != null) { + next.setHeadValue(carrier.get(next.getHeadKey())); + } + } + } + + SpanRef span = Tracer.createEntrySpan(functionName, contextCarrierRef); + if (tags != null) { + for (String key : tags.keySet()) { + span.tag(key, tags.get(key)); + } + } + + if (baggage != null) { + for (String key : baggage.keySet()) { + TraceContext.putCorrelation(key, baggage.get(key)); + } + } + + Error err = callback.execute(); + if (err != null) { + ActiveSpan.error(err); + } + Tracer.stopSpan(); } @Override public void executeWithTracing(Callback callback) throws Exception { - + executeWithTracing("function", null, callback); } @Override public void executeWithTracing(Plugin plugin, Callback callback) throws Exception { - + Map tags = new HashMap<>(); + tags.put("kind", "Plugin"); + tags.put("name", plugin.name()); + tags.put("version", plugin.version()); + if (plugin.tagsAddToTracing() != null) { + tags.putAll(plugin.tagsAddToTracing()); + } + + executeWithTracing(plugin.name(), tags, callback); } @Override public void executeWithTracing(Hook hook, Callback callback) throws Exception { - + Map tags = new HashMap<>(); + tags.put("kind", "Hook"); + tags.put("name", hook.name()); + tags.put("version", hook.version()); + if (hook.tagsAddToTracing() != null) { + tags.putAll(hook.tagsAddToTracing()); + } + + executeWithTracing(hook.name(), tags, callback); } @Override public void executeWithTracing(UserContext ctx, Callback callback) throws Exception { + Map tags = new HashMap<>(); + tags.put("function", ctx.getFunctionClass().getName()); + + executeWithTracing(ctx.getFunctionClass().getSimpleName(), tags, callback); + } + + private void executeWithTracing(String name, Map tags, Callback callback) throws Exception { + SpanRef span = Tracer.createLocalSpan(name); + + if (tags != null) { + for (String key : tags.keySet()) { + span.tag(key, tags.get(key)); + } + } + + Error err = callback.execute(); + if (err != null) { + ActiveSpan.error(err); + } + Tracer.stopSpan(); } } diff --git a/functions-framework-invoker/src/main/resources/nocalhost/debug.sh b/functions-framework-invoker/src/main/resources/nocalhost/debug.sh index 1137002e..0ee86a14 100755 --- a/functions-framework-invoker/src/main/resources/nocalhost/debug.sh +++ b/functions-framework-invoker/src/main/resources/nocalhost/debug.sh @@ -1,4 +1,4 @@ #!/bin/bash mvn clean compile dependency:copy-dependencies -mvn exec:exec -Dexec.executable="java" -Dexec.args="-classpath samples-1.0-SNAPSHOT.jar:target/classes/:target/dependency/* -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5005 dev.openfunction.invoker.Runner" +mvn exec:exec -Dexec.executable="java" -Dexec.args="-javaagent:./skywalking-agent/skywalking-agent.jar -classpath samples-1.0-SNAPSHOT.jar:target/classes/:target/dependency/* -Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5005 dev.openfunction.invoker.Runner" diff --git a/functions-framework-invoker/src/main/resources/nocalhost/run.sh b/functions-framework-invoker/src/main/resources/nocalhost/run.sh index 04590525..34db50ab 100755 --- a/functions-framework-invoker/src/main/resources/nocalhost/run.sh +++ b/functions-framework-invoker/src/main/resources/nocalhost/run.sh @@ -1,4 +1,4 @@ #!/bin/bash mvn clean compile dependency:copy-dependencies -mvn exec:exec -Dexec.executable="java" -Dexec.args="-classpath samples-1.0-SNAPSHOT.jar:target/classes/:target/dependency/* dev.openfunction.invoker.Runner" +mvn exec:exec -Dexec.executable="java" -Dexec.args="-javaagent:./skywalking-agent/skywalking-agent.jar -classpath samples-1.0-SNAPSHOT.jar:target/classes/:target/dependency/* dev.openfunction.invoker.Runner" diff --git a/samples/pom.xml b/samples/pom.xml index 031f8660..a061d41d 100644 --- a/samples/pom.xml +++ b/samples/pom.xml @@ -31,7 +31,7 @@ dev.openfunction.functions functions-framework-api - 1.2.0-SNAPSHOT + 1.3.0-SNAPSHOT diff --git a/samples/src/main/java/dev/openfunction/samples/OpenFunctionImpl.java b/samples/src/main/java/dev/openfunction/samples/OpenFunctionImpl.java index 8c685ab3..1941e11f 100644 --- a/samples/src/main/java/dev/openfunction/samples/OpenFunctionImpl.java +++ b/samples/src/main/java/dev/openfunction/samples/OpenFunctionImpl.java @@ -37,10 +37,12 @@ public Out accept(Context context, String payload) throws Exception { for (String key : context.getOutputs().keySet()) { Component output = context.getOutputs().get(key); if (output.isPubsub()) { - daprClient.publishEvent(output.getComponentName(), output.getTopic(), payload, output.getMetadata()); + daprClient.publishEvent(output.getComponentName(), output.getTopic(), payload, output.getMetadata()).block(); + System.out.println("send data to " + output.getComponentName()); } else if (output.isBinding()) { // We recommend using CloudEvents to pass data between Dapr components. - daprClient.invokeBinding(output.getComponentName(), output.getOperation(), context.packageAsCloudevent(payload)); + daprClient.invokeBinding(output.getComponentName(), output.getOperation(), context.packageAsCloudevent(payload), output.getMetadata()).block(); + System.out.println("send data to " + output.getComponentName()); } } }