diff --git a/functions-framework-api/pom.xml b/functions-framework-api/pom.xml index 26b466a8..9e68a03b 100644 --- a/functions-framework-api/pom.xml +++ b/functions-framework-api/pom.xml @@ -28,6 +28,11 @@ dapr-sdk 1.8.0 + + org.apache.eventmesh + eventmesh-sdk-java + 1.9.0-release + @@ -38,7 +43,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..071b4971 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 @@ -16,6 +16,8 @@ package dev.openfunction.functions; +import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer; + import io.cloudevents.CloudEvent; import io.dapr.client.DaprClient; @@ -121,5 +123,7 @@ public interface Context { */ DaprClient getDaprClient(); - CloudEvent packageAsCloudevent(String payload); + EventMeshGrpcProducer getEventMeshProducer(); + + byte[] packageAsCloudevent(String payload); } diff --git a/functions-framework-invoker/pom.xml b/functions-framework-invoker/pom.xml index 35a70084..516f88a8 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,17 @@ dev.openfunction.functions functions-framework-api - 1.2.0-SNAPSHOT + 1.3.0-SNAPSHOT + + + org.apache.eventmesh + eventmesh-connector-openfunction + 1.9.0-release + + + org.apache.eventmesh + eventmesh-sdk-java + 1.9.0-release io.cloudevents @@ -72,6 +83,11 @@ cloudevents-core ${cloudevents.sdk.version} + + io.cloudevents + cloudevents-protobuf + ${cloudevents.sdk.version} + com.google.auto.value auto-value @@ -179,6 +195,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/Runner.java b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/Runner.java index efb73116..004d24dd 100644 --- a/functions-framework-invoker/src/main/java/dev/openfunction/invoker/Runner.java +++ b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/Runner.java @@ -19,6 +19,7 @@ import dev.openfunction.invoker.context.RuntimeContext; import dev.openfunction.invoker.trigger.DaprTrigger; +import dev.openfunction.invoker.trigger.EventMeshTrigger; import dev.openfunction.invoker.trigger.HttpTrigger; import dev.openfunction.invoker.trigger.Trigger; import org.apache.commons.lang3.StringUtils; @@ -82,6 +83,10 @@ public static void main(String[] args) { triggers.add(new DaprTrigger(runtimeContext, functionClasses)); } + if (runtimeContext.hasEventMeshTrigger()) { + triggers.add(new EventMeshTrigger(runtimeContext, functionClasses)); + } + for (Trigger trigger : triggers) { trigger.start(); } 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..66138cb8 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; } @@ -182,6 +172,8 @@ static class Triggers { private HttpTrigger http; private DaprTrigger[] dapr; + private EventMeshTrigger eventMesh; + public HttpTrigger getHttp() { return http; } @@ -197,6 +189,14 @@ public DaprTrigger[] getDapr() { public void setDapr(DaprTrigger[] dapr) { this.dapr = dapr; } + + public EventMeshTrigger getEventMesh() { + return eventMesh; + } + + public void setEventMesh(EventMeshTrigger eventMesh) { + this.eventMesh = eventMesh; + } } static class HttpTrigger { @@ -250,4 +250,99 @@ public void setInputName(String inputName) { this.inputName = inputName; } } + + static class EventMeshTrigger { + private String name; + + private String type; + + private String topic; + + private String eventMeshConnectorAddr; + + private String eventMeshConnectorPort; + + private String producerGroup; + + private String env; + + private String idc; + + private String sysId; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + + public String getEventMeshConnectorAddr() { + return eventMeshConnectorAddr; + } + + public void setEventMeshConnectorAddr(String eventMeshConnectorAddr) { + this.eventMeshConnectorAddr = eventMeshConnectorAddr; + } + + public String getEventMeshConnectorPort() { + return eventMeshConnectorPort; + } + + public void setEventMeshConnectorPort(String eventMeshConnectorPort) { + this.eventMeshConnectorPort = eventMeshConnectorPort; + } + + public String getProducerGroup() { + return producerGroup; + } + + public void setProducerGroup(String producerGroup) { + this.producerGroup = producerGroup; + } + + public String getEnv() { + return env; + } + + public void setEnv(String env) { + this.env = env; + } + + public String getIdc() { + return idc; + } + + public void setIdc(String idc) { + this.idc = idc; + } + + public String getSysId() { + return sysId; + } + + public void setSysId(String sysId) { + this.sysId = sysId; + } + } + + } 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..f153ed45 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 @@ -26,7 +26,7 @@ import dev.openfunction.invoker.tracing.TracingProvider; import io.cloudevents.CloudEvent; import io.cloudevents.core.provider.EventFormatProvider; -import org.apache.commons.collections.CollectionUtils; + import org.apache.commons.collections.MapUtils; import org.apache.commons.lang3.ArrayUtils; @@ -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)) { @@ -212,6 +227,36 @@ public Map getDaprTrigger() { return null; } + + public boolean hasEventMeshTrigger() { + return functionContext.getTriggers() != null && functionContext.getTriggers().getEventMesh() != null; + } + + public Component getEventMeshTrigger() { + if (functionContext.getTriggers() != null && + functionContext.getTriggers().getEventMesh() != null) { + FunctionContext.EventMeshTrigger trigger = functionContext.getTriggers().getEventMesh(); + + Component component = new Component(); + component.setComponentName(trigger.getName()); + component.setComponentType(trigger.getType()); + component.setTopic(trigger.getTopic()); + Map metaDataMap = new HashMap<>(); + metaDataMap.put("eventMeshConnectorAddr", trigger.getEventMeshConnectorAddr()); + metaDataMap.put("eventMeshConnectorPort", trigger.getEventMeshConnectorPort()); + metaDataMap.put("producerGroup", trigger.getProducerGroup()); + metaDataMap.put("env", trigger.getEnv()); + metaDataMap.put("idc", trigger.getIdc()); + metaDataMap.put("sysId", trigger.getSysId()); + + component.setMetadata(metaDataMap); + + return component; + } + + return null; + } + public FunctionContext getFunctionContext() { return functionContext; } 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..82521bae 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 @@ -24,6 +24,8 @@ import io.dapr.client.DaprClient; import jakarta.servlet.http.HttpServletResponse; +import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer; + import java.net.URI; import java.util.*; import java.util.logging.Level; @@ -44,7 +46,9 @@ public class UserContext implements Context { ); private final RuntimeContext runtimeContext; - private final DaprClient daprClient; + private DaprClient daprClient; + + private EventMeshGrpcProducer eventMeshGrpcProducer; private Out out; @@ -62,6 +66,11 @@ public UserContext(RuntimeContext runtimeContext, DaprClient daprClient) { this.daprClient = daprClient; } + public UserContext(RuntimeContext runtimeContext, EventMeshGrpcProducer eventMeshGrpcProducer) { + this.runtimeContext = runtimeContext; + this.eventMeshGrpcProducer = eventMeshGrpcProducer; + } + public UserContext withHttp(HttpRequest httpRequest, HttpResponse httpResponse) { this.httpRequest = httpRequest; this.httpResponse = httpResponse; @@ -101,8 +110,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 +177,20 @@ public DaprClient getDaprClient() { } @Override - public CloudEvent packageAsCloudevent(String payload) { - return new CloudEventBuilder() + public EventMeshGrpcProducer getEventMeshProducer() { + return eventMeshGrpcProducer; + } + + @Override + 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/java/dev/openfunction/invoker/trigger/EventMeshTrigger.java b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/trigger/EventMeshTrigger.java new file mode 100644 index 00000000..d7433d98 --- /dev/null +++ b/functions-framework-invoker/src/main/java/dev/openfunction/invoker/trigger/EventMeshTrigger.java @@ -0,0 +1,162 @@ +/* +Copyright 2022 The OpenFunction Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package dev.openfunction.invoker.trigger; + +import org.apache.commons.collections.MapUtils; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Map; +import java.util.Objects; +import java.util.logging.Level; +import java.util.logging.Logger; + +import io.cloudevents.core.format.EventFormat; +import io.cloudevents.core.provider.EventFormatProvider; +import io.cloudevents.protobuf.ProtobufFormat; +import io.grpc.Server; +import io.grpc.ServerBuilder; +import io.grpc.stub.StreamObserver; + +import dev.openfunction.functions.CloudEventFunction; +import dev.openfunction.functions.Component; +import dev.openfunction.functions.OpenFunction; +import dev.openfunction.invoker.context.RuntimeContext; +import dev.openfunction.invoker.context.UserContext; + +import org.apache.eventmesh.client.grpc.config.EventMeshGrpcClientConfig; +import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer; +import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent; +import org.apache.eventmesh.connector.openfunction.client.CallbackServiceGrpc; + +/** + * Executes the user's asynchronous function. + */ +public final class EventMeshTrigger implements Trigger { + + private static final Logger logger = Logger.getLogger("dev.openfunction.invoker"); + + private static final EventFormat eventFormat = EventFormatProvider.getInstance().resolveFormat(ProtobufFormat.PROTO_CONTENT_TYPE); + + private final RuntimeContext runtimeContext; + + private final ArrayList functions; + + private final Service service; + + public EventMeshTrigger(RuntimeContext runtimeContext, Class[] functionClasses) { + this.runtimeContext = runtimeContext; + + functions = new ArrayList<>(); + for (Class c : functionClasses) { + try { + if (OpenFunction.class.isAssignableFrom(c)) { + Class openFunctionClass = c.asSubclass(OpenFunction.class); + functions.add(openFunctionClass.getConstructor().newInstance()); + } else { + throw new Error("Unsupported function " + c.getName()); + } + + } catch (ReflectiveOperationException e) { + throw new Error("Could not construct an instance of " + c.getName(), e); + } + } + + service = new Service(); + } + + @Override + public void start() throws Exception { + if (runtimeContext.getEventMeshTrigger() == null) { + throw new Error("no eventmesh trigger defined for the function"); + } + + this.service.start(runtimeContext.getPort()); + } + + @Override + public void close() { + + } + + private class Service extends CallbackServiceGrpc.CallbackServiceImplBase { + + private Server eventMeshServer; + + private EventMeshGrpcProducer eventMeshGrpcProducer; + + public void start(int port) throws Exception { + eventMeshServer = ServerBuilder + .forPort(port) + .addService(Service.this) + .build() + .start(); + + // this map only have one trigger + Component eventMeshTrigger = runtimeContext.getEventMeshTrigger(); + Map metaDataMap = eventMeshTrigger.getMetadata(); + EventMeshGrpcClientConfig eventMeshGrpcClientConfig = EventMeshGrpcClientConfig.builder() + .serverAddr(metaDataMap.get("eventMeshConnectorAddr")) + .serverPort(Integer.parseInt(metaDataMap.get("eventMeshConnectorPort"))) + .producerGroup(metaDataMap.get("producerGroup")) + .env(metaDataMap.get("env")) + .idc(metaDataMap.get("idc")) + .sys(metaDataMap.get("sysId")) + .build(); + + eventMeshGrpcProducer = new EventMeshGrpcProducer(eventMeshGrpcClientConfig); + + // Now we handle ctrl+c (or any other JVM shutdown) + Runtime.getRuntime().addShutdownHook(new Thread(() -> { + eventMeshGrpcProducer.close(); + eventMeshServer.shutdown(); + }) + ); + + eventMeshServer.awaitTermination(); + } + + @Override + public void onTopicEvent(CloudEvent cloudEvent, StreamObserver responseObserver) { + + assert eventFormat != null; + io.cloudevents.CloudEvent event = eventFormat.deserialize(cloudEvent.toByteArray()); + + try { + runtimeContext.executeWithTracing(event, () -> { + for (Object function : functions) { + runtimeContext.executeWithTracing(event, () -> { + new UserContext(runtimeContext, eventMeshGrpcProducer). + executeFunction((OpenFunction) function, + new String(Objects.requireNonNull(event.getData()).toBytes(), StandardCharsets.UTF_8)); + return null; + } + ); + } + responseObserver.onNext(CloudEvent.getDefaultInstance()); + responseObserver.onCompleted(); + return null; + } + ); + } catch (Exception e) { + logger.log(Level.INFO, "catch exception when execute function " + runtimeContext.getName()); + e.printStackTrace(); + responseObserver.onError(e); + } + } + } +} 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..369a6307 100644 --- a/samples/pom.xml +++ b/samples/pom.xml @@ -31,7 +31,17 @@ dev.openfunction.functions functions-framework-api - 1.2.0-SNAPSHOT + 1.3.0-SNAPSHOT + + + io.dapr + dapr-sdk + 1.8.0 + + + org.apache.eventmesh + eventmesh-sdk-java + 1.9.0-release diff --git a/samples/src/main/java/dev/openfunction/samples/OpenFunctionEventMeshImpl.java b/samples/src/main/java/dev/openfunction/samples/OpenFunctionEventMeshImpl.java new file mode 100644 index 00000000..fcb7b7e5 --- /dev/null +++ b/samples/src/main/java/dev/openfunction/samples/OpenFunctionEventMeshImpl.java @@ -0,0 +1,68 @@ +/* +Copyright 2022 The OpenFunction Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package dev.openfunction.samples; + +import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer; +import org.apache.eventmesh.client.tcp.common.EventMeshCommon; +import org.apache.eventmesh.common.Constants; +import org.apache.eventmesh.common.utils.JsonUtils; + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.UUID; + +import io.cloudevents.CloudEvent; +import io.cloudevents.core.builder.CloudEventBuilder; + +import dev.openfunction.functions.Component; +import dev.openfunction.functions.Context; +import dev.openfunction.functions.OpenFunction; +import dev.openfunction.functions.Out; + +public class OpenFunctionEventMeshImpl implements OpenFunction { + + @Override + public Out accept(Context context, String payload) throws Exception { + System.out.printf("receive event: %s", payload).println(); + + EventMeshGrpcProducer eventMeshGrpcProducer = context.getEventMeshProducer(); + if (eventMeshGrpcProducer == null) { + return new Out(); + } + + if (context.getOutputs() != null) { + for (String key : context.getOutputs().keySet()) { + Component output = context.getOutputs().get(key); + eventMeshGrpcProducer.publish(buildCloudEvent(output, payload)); + } + } + return new Out(); + } + + protected static CloudEvent buildCloudEvent(Component output, String content) { + return CloudEventBuilder.v1() + .withId(UUID.randomUUID().toString()) + .withSubject(output.getTopic()) + .withSource(URI.create("/")) + .withDataContentType("application/cloudevents+json") + .withType(EventMeshCommon.CLOUD_EVENTS_PROTOCOL_NAME) + .withData(JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8)) + .withExtension(Constants.EVENTMESH_MESSAGE_CONST_TTL, String.valueOf(4 * 1000)) + .build(); + + } +} 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()); } } }