Integration Apache SkyWalking (#30)

.github/workflows/plugin_test.yaml

@@ -0,0 +1,27 @@
+name: PluginsTest
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  Plugins-Test:
+    name: SkyWalking
+    runs-on: ubuntu-latest
+    env:
+      e2e_file: "plugin/skywalking/test/sync-request/e2e.yaml"
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          submodules: true
+      - uses: apache/skywalking-infra-e2e@main
+        with:
+          e2e-file: ${e2e_file}
+      - name: Show Container Logs
+        if: ${{ failure() }}
+        run: docker ps -a | grep -v CONTAINER | awk '{print $1}' | xargs -i docker logs {}
+      - name: Cleanup
+        if: ${{ failure() }}
+        run: e2e cleanup -c ${e2e_file}

go.mod

@@ -3,8 +3,10 @@ module github.com/OpenFunction/functions-framework-go
 go 1.15
 
 require (
+	github.com/SkyAPM/go2sky v1.4.0
 	github.com/cloudevents/sdk-go/v2 v2.4.1
 	github.com/dapr/go-sdk v1.2.0
 	github.com/fatih/structs v1.1.0
 	k8s.io/klog/v2 v2.30.0
+	skywalking.apache.org/repo/goapi v0.0.0-20220121092418-9c455d0dda3f
 )

plugin/skywalking/plugin-skywalking.go

@@ -0,0 +1,129 @@
+package skywalking
+
+import (
+	"context"
+	"sync"
+
+	ofctx "github.com/OpenFunction/functions-framework-go/context"
+	"github.com/OpenFunction/functions-framework-go/plugin"
+	"github.com/SkyAPM/go2sky"
+	"github.com/SkyAPM/go2sky/reporter"
+	"k8s.io/klog/v2"
+
+	agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
+)
+
+const (
+	name    = "skywalking"
+	version = "v1"
+
+	componentIDOpenFunction = 5013 // https://github.com/apache/skywalking/blob/master/oap-server/server-starter/src/main/resources/component-libraries.yml#L515
+)
+
+var (
+	initGo2skyOnce sync.Once
+)
+
+type klogWrapper struct {
+}
+
+func (k klogWrapper) Info(args ...interface{}) {
+	klog.Info(args)
+}
+
+func (k klogWrapper) Infof(format string, args ...interface{}) {
+	klog.Infof(format, args)
+}
+
+func (k klogWrapper) Warn(args ...interface{}) {
+	klog.Warning(args)
+}
+
+func (k klogWrapper) Warnf(format string, args ...interface{}) {
+	klog.Warningf(format, args)
+}
+
+func (k klogWrapper) Error(args ...interface{}) {
+	klog.Error(args)
+}
+
+func (k klogWrapper) Errorf(format string, args ...interface{}) {
+	klog.Errorf(format, args)
+}
+
+func initGo2sky(ofCtx ofctx.Context, p *PluginSkywalking) {
+	initGo2skyOnce.Do(func() {
+		r, err := reporter.NewGRPCReporter(ofCtx.PluginsTracing.Provider.OapServer, reporter.WithLog(&klogWrapper{}))
+		if err != nil {
+			klog.Errorf("new go2sky grpc reporter error\n", err)
+			return
+		}
+		tracer, err := go2sky.NewTracer(ofCtx.Name, go2sky.WithReporter(r), go2sky.WithInstance(ofCtx.PluginsTracing.Tags["instance"]))
+		if err != nil {
+			klog.Errorf("new go2sky tracer error\n", err)
+			return
+		}
+		go2sky.SetGlobalTracer(tracer)
+
+		p.tracer = tracer
+	})
+}
+
+type PluginSkywalking struct {
+	tracer *go2sky.Tracer
+}
+
+func (p *PluginSkywalking) Init() plugin.Plugin {
+	return p
+}
+
+func (p PluginSkywalking) Name() string {
+	return name
+}
+
+func (p PluginSkywalking) Version() string {
+	return version
+
+}
+
+func (p *PluginSkywalking) ExecPreHook(ctx ofctx.Context, plugins map[string]plugin.Plugin) error {
+	initGo2sky(ctx, p)
+	if p.tracer == nil {
+		return nil
+	}
+
+	if ctx.SyncRequestMeta.Request != nil {
+		// SyncRequest
+		return preSyncRequestLogic(&ctx, p.tracer)
+	}
+	return nil
+}
+
+func (p *PluginSkywalking) ExecPostHook(ctx ofctx.Context, plugins map[string]plugin.Plugin) error {
+	if p.tracer == nil {
+		return nil
+	}
+	if ctx.SyncRequestMeta.Request != nil {
+		return postSyncRequestLogic(&ctx)
+	}
+	return nil
+}
+
+func (p PluginSkywalking) Get(fieldName string) (interface{}, bool) {
+	return nil, false
+}
+
+func setPublicAttrs(ctx context.Context, ofCtx *ofctx.Context, span go2sky.Span) {
+	span.SetSpanLayer(agentv3.SpanLayer_FAAS)
+	span.SetComponent(componentIDOpenFunction)
+
+	// tags
+	for key, value := range ofCtx.PluginsTracing.Tags {
+		span.Tag(go2sky.Tag(key), value)
+	}
+	// baggage
+	for key, value := range ofCtx.PluginsTracing.Baggage {
+		go2sky.PutCorrelation(ctx, key, value)
+	}
+
+}

plugin/skywalking/sync-request.go

@@ -0,0 +1,44 @@
+package skywalking
+
+import (
+	"fmt"
+	"time"
+
+	ofctx "github.com/OpenFunction/functions-framework-go/context"
+	"github.com/SkyAPM/go2sky"
+)
+
+func preSyncRequestLogic(ofCtx *ofctx.Context, tracer *go2sky.Tracer) error {
+	request := ofCtx.SyncRequestMeta.Request
+
+	span, nCtx, err := tracer.CreateEntrySpan(ofCtx.SyncRequestMeta.Request.Context(), ofCtx.Name, func(key string) (string, error) {
+		return request.Header.Get(key), nil
+	})
+	if err != nil {
+		return err
+	}
+	ofCtx.SyncRequestMeta.Request = request.WithContext(nCtx)
+
+	span.Tag(go2sky.TagHTTPMethod, request.Method)
+	span.Tag(go2sky.TagURL, fmt.Sprintf("%s%s", request.Host, request.URL.Path))
+	setPublicAttrs(nCtx, ofCtx, span)
+	return nil
+}
+
+func postSyncRequestLogic(ctx *ofctx.Context) error {
+	request := ctx.SyncRequestMeta.Request
+	span := go2sky.ActiveSpan(request.Context())
+	if span == nil {
+		return nil
+	}
+	defer span.End()
+
+	if ofctx.InternalError == ctx.Out.Code {
+		span.Error(time.Now(), "Error on handling request")
+	}
+
+	if ctx.Error != nil {
+		span.Error(time.Now(), ctx.Error.Error())
+	}
+	return nil
+}

plugin/skywalking/test/sync-request/docker-compose.yml

@@ -0,0 +1,92 @@
+version: '2.1'
+
+services:
+  mockoap:
+    image: ghcr.io/apache/skywalking-agent-test-tool/mock-collector:3792b52c5b27462630bfd556bdc4cc7b5c41b321
+    expose:
+      - 19876
+      - 12800
+    ports:
+      - 19876
+      - 12800
+    networks:
+      - e2e
+    restart: on-failure
+    healthcheck:
+      test: [ "CMD", "curl", "http://127.0.0.1:12800/healthCheck" ]
+      interval: 5s
+      timeout: 60s
+      retries: 120
+  provider:
+    build:
+      context: ../../../../
+      dockerfile: ./plugin/skywalking/test/sync-request/docker/Dockerfile.provider
+    environment:
+      SW_AGENT_COLLECTOR_GET_AGENT_DYNAMIC_CONFIG_INTERVAL: -1
+      SW_AGENT_COLLECTOR_BACKEND_SERVICES: mockoap:19876
+    networks:
+      - e2e
+    expose:
+      - 8080
+    ports:
+      - 8080
+    healthcheck:
+      test: ["CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/8080"]
+      interval: 5s
+      timeout: 60s
+      retries: 120
+    depends_on:
+      mockoap:
+        condition: service_healthy
+  of:
+    build:
+      context: ../../../../
+      dockerfile: ./plugin/skywalking/test/sync-request/docker/Dockerfile.of
+    environment:
+      SW_AGENT_COLLECTOR_GET_AGENT_DYNAMIC_CONFIG_INTERVAL: -1
+      FUNC_CONTEXT: |
+        {
+          "name": "function-test",
+          "version": "v1.0.0",
+          "runtime": "Knative",
+          "port": "12345",
+          "prePlugins": [],
+          "postPlugins": [],
+          "pluginsTracing": {
+            "enable": true,
+            "provider": {
+              "name": "skywalking",
+              "oapServer": "mockoap:19876"
+            },
+            "tags": {
+              "func": "function-test",
+              "layer": "faas",
+              "tag1": "value1",
+              "tag2": "value2"
+            },
+            "baggage": {
+              "CONSUMER_KEY": "of"
+            }
+          }
+        }
+      POD_NAME: function-test-vhct4
+      POD_NAMESPACE: test
+      PROVIDER_ADDRESS: http://provider:8080
+    expose:
+      - 12345
+    ports:
+      - 12345
+    networks:
+      - e2e
+    healthcheck:
+      test: ["CMD", "bash", "-c", "cat < /dev/null > /dev/tcp/127.0.0.1/8080"]
+      interval: 5s
+      timeout: 60s
+      retries: 120
+    depends_on:
+      mockoap:
+        condition: service_healthy
+      provider:
+        condition: service_healthy
+networks:
+  e2e:

plugin/skywalking/test/sync-request/docker/Dockerfile.of

@@ -0,0 +1,12 @@
+FROM golang:1.16
+
+ADD . /functions-framework-go
+WORKDIR /functions-framework-go
+
+ENV GOPROXY="https://goproxy.cn"
+
+EXPOSE 12345
+
+ENTRYPOINT ["go"]
+
+CMD ["run", "plugin/skywalking/test/sync-request/of/of.go"]

plugin/skywalking/test/sync-request/docker/Dockerfile.provider

@@ -0,0 +1,12 @@
+FROM golang:1.16
+
+ADD . /functions-framework-go
+WORKDIR /functions-framework-go
+
+ENV GOPROXY="https://goproxy.cn"
+
+EXPOSE 12345
+
+ENTRYPOINT ["go"]
+
+CMD ["run", "plugin/skywalking/test/sync-request/provider/provider.go"]

plugin/skywalking/test/sync-request/e2e.yaml

@@ -0,0 +1,26 @@
+setup:
+  env: compose
+  file: docker-compose.yml
+  timeout: 20m
+
+cleanup:
+  # always never success failure
+  on: success
+
+trigger:
+  action: http
+  interval: 3s
+  times: 5
+  url: http://${of_host}:${of_12345}
+  method: GET
+
+verify:
+  # verify with retry strategy
+  retry:
+    # max retry count
+    count: 10
+    # the interval between two retries, in millisecond.
+    interval: 10s
+  cases:
+    - query: curl http://${mockoap_host}:${mockoap_12800}/receiveData
+      expected: expected.data.yml