Streaming events from pubsub to the HTTP webhook

Co-Authored-By: Peter Hazell <[email protected]> Co-Authored-By: Alex Georgousis <[email protected]> Co-Authored-By: Toby Pinfold <[email protected]> Co-Authored-By: Ian <[email protected]>
sky-uk · Nov 8, 2024 · fcd4ed7 · fcd4ed7
1 parent 5e41f9d
commit fcd4ed7
Show file tree

Hide file tree

Showing 31 changed files with 4,232 additions and 142 deletions.
diff --git a/Makefile b/Makefile
@@ -228,6 +228,12 @@ docker-build-triggers:
 docker-push-triggers:
 	$(MAKE) -C triggers/run-completion-event-trigger docker-push
 
+docker-build-providers:
+	$(MAKE) -C provider-service docker-build
+
+docker-push-providers:
+	$(MAKE) -C provider-service docker-push
+
 ##@ Docs
 
 website:
@@ -238,9 +244,9 @@ docker-push-quickstart:
 
 ##@ Package
 
-package-all: docker-build docker-build-argo docker-build-triggers helm-package website
+package-all: docker-build docker-build-argo docker-build-triggers docker-build-providers helm-package website
 
-publish-all: docker-push docker-push-argo docker-push-triggers helm-publish
+publish-all: docker-push docker-push-argo docker-push-triggers docker-push-providers helm-publish
 
 ##@ CI
 

diff --git a/controllers/webhook/runcompletion_feed.go b/controllers/webhook/runcompletion_feed.go
@@ -62,11 +62,8 @@ func getRequestBody(ctx context.Context, request *http.Request) ([]byte, error)
 	} else if len(body) == 0 {
 		return nil, errors.New("request body is empty")
 	}
-	return body, nil
-}
 
-type DataWrapper struct {
-	Data common.RunCompletionEventData `json:"data"`
+	return body, nil
 }
 
 func (rcf RunCompletionFeed) extractRunCompletionEvent(request *http.Request) (*common.RunCompletionEvent, error) {
@@ -75,12 +72,12 @@ func (rcf RunCompletionFeed) extractRunCompletionEvent(request *http.Request) (*
 		return nil, err
 	}
 
-	runDataWrapper := &DataWrapper{}
-	if err := json.Unmarshal(body, runDataWrapper); err != nil {
+	rced := common.RunCompletionEventData{}
+	if err := json.Unmarshal(body, &rced); err != nil {
 		return nil, err
 	}
 
-	rce, err := rcf.eventProcessor.ToRunCompletionEvent(rcf.ctx, runDataWrapper.Data)
+	rce, err := rcf.eventProcessor.ToRunCompletionEvent(rcf.ctx, rced)
 	if err != nil {
 		return nil, err
 	} else if rce == nil {
@@ -95,7 +92,7 @@ func (rcf RunCompletionFeed) handleEvent(response http.ResponseWriter, request *
 	switch request.Method {
 	case http.MethodPost:
 		if request.Header.Get(HttpHeaderContentType) != HttpContentTypeJSON {
-			logger.Error(errors.New("RunCompletionFeed call failed"), "invalid %s [%s], want `%s`", HttpHeaderContentType, request.Header.Get(HttpHeaderContentType), HttpContentTypeJSON)
+			logger.Error(errors.New("RunCompletionFeed call failed"), fmt.Sprintf("invalid %s [%s], want `%s`", HttpHeaderContentType, request.Header.Get(HttpHeaderContentType), HttpContentTypeJSON))
 			http.Error(response, fmt.Sprintf("invalid %s, want `%s`", HttpHeaderContentType, HttpContentTypeJSON), http.StatusUnsupportedMediaType)
 			return
 		}

diff --git a/go.mod b/go.mod
@@ -78,6 +78,7 @@ require (
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/reugn/go-streams v0.10.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect

diff --git a/go.sum b/go.sum
@@ -468,6 +468,8 @@ github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1
 github.com/prometheus/procfs v0.7.3 h1:4jVXhlkAyzOScmCkXBTOLRLTz8EeU+eyjrwB/EPq0VU=
 github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/reugn/go-streams v0.10.0 h1:Y0wHNihEbHsFOFV2/xTOKvud4ZpJPaRTET01fwx2/rQ=
+github.com/reugn/go-streams v0.10.0/go.mod h1:QI5XXifJkVJl2jQ6Cra8I9DvWdJTgqcFYR7amvXZ9Lg=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=

diff --git a/helm/provider/templates/_helpers.tpl b/helm/provider/templates/_helpers.tpl
@@ -30,6 +30,16 @@ Prefix for provider resources, allows for override with .Values.prefixOverride
 {{/*
 Generate docker image location for the provider
 */}}
+{{/*
+Provider CLI Image
+*/}}
 {{- define "kfp-operator-provider.image" -}}
 {{ include "kfp-operator-provider.trimmedContainerRegistry" . }}kfp-operator-{{ .Values.provider.type }}-provider:{{ .Chart.AppVersion }}
 {{- end }}
+
+{{/*
+Provider Deployment Image
+*/}}
+{{- define "kfp-operator-provider-service.image" -}}
+{{ include "kfp-operator-provider.trimmedContainerRegistry" . }}kfp-operator-provider-service:{{ .Chart.AppVersion }}
+{{- end }}
diff --git a/helm/provider/templates/runcompletion/_helpers.tpl b/helm/provider/templates/runcompletion/_helpers.tpl
diff --git a/helm/provider/templates/runcompletion/eventsource.yaml b/helm/provider/templates/runcompletion/eventsource.yaml
diff --git a/helm/provider/templates/runcompletion/eventsource_server.yaml b/helm/provider/templates/runcompletion/eventsource_server.yaml
diff --git a/helm/provider/templates/runcompletion/sensor.yaml b/helm/provider/templates/runcompletion/sensor.yaml
diff --git a/helm/provider/templates/service/_helpers.tpl b/helm/provider/templates/service/_helpers.tpl
@@ -0,0 +1,7 @@
+{{/*
+Common labels
+*/}}
+{{- define "kfp-operator-provider.service.labels" -}}
+app: {{ include "kfp-operator-provider.resource-prefix" . }}-service
+provider: {{ .Values.provider.name }}
+{{- end }}
diff --git a/helm/provider/templates/service/config.yaml b/helm/provider/templates/service/config.yaml
@@ -0,0 +1,12 @@
+{{- $providerName := .Values.provider.name }}
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ include "kfp-operator-provider.resource-prefix" $ }}-service-config
+  namespace: {{ include "kfp-operator-provider.namespace" $ }}
+data:
+  config.yaml: |
+    providerName: {{ $providerName }}
+    operatorWebhook: http://{{ .Values.kfpOperator.fullname}}-runcompletion-webhook-service/events
+    pod:
+      namespace: to-be-overridden
diff --git a/helm/provider/templates/service/provider-service.yaml b/helm/provider/templates/service/provider-service.yaml
@@ -0,0 +1,41 @@
+{{- $providerName := .Values.provider.name }}
+{{- $labels := include "kfp-operator-provider.service.labels" $ }}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ include "kfp-operator-provider.resource-prefix" $ }}-{{ $providerName }}-service
+  namespace: {{ include "kfp-operator-provider.namespace" $ }}
+  labels:
+    {{- $labels | nindent 4 }}
+spec:
+  selector:
+    matchLabels:
+      {{- $labels | nindent 6 }}
+  replicas: 1
+  template:
+    metadata:
+      {{- (deepCopy $.Values.service.metadata | merge (dict "labels" ($labels | fromYaml))) | toYaml | nindent 6 }}
+    spec:
+      serviceAccountName: {{ .Values.service.serviceAccount.name }}
+      securityContext:
+        runAsNonRoot: true
+      containers:
+      - name: provider-service
+        image: {{ include "kfp-operator-provider-service.image" $ }}
+        securityContext:
+          allowPrivilegeEscalation: false
+        resources:
+          {{- $.Values.service.resources | toYaml | nindent 10 }}
+        env:
+        - name: POD_NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        volumeMounts:
+          - name: config
+            mountPath: /etc/provider-service/config.yaml
+            subPath: config.yaml
+      volumes:
+        - name: config
+          configMap:
+            name: {{ include "kfp-operator-provider.resource-prefix" . }}-service-config
diff --git a/helm/provider/values.yaml b/helm/provider/values.yaml
@@ -28,12 +28,10 @@ provider:
 logging:
   verbosity: # info
 
-eventsourceServer:
+service:
   serviceAccount:
     name: kfp-operator-kfp
-  create: false
   metadata: {}
-  port: 8080
   resources:
     limits:
       cpu: 100m

diff --git a/provider-service/Dockerfile b/provider-service/Dockerfile
@@ -0,0 +1,7 @@
+FROM gcr.io/distroless/static-debian12:nonroot
+
+COPY bin/provider-service /bin/provider-service
+
+COPY internal/config/config.yaml /etc/provider-service/config.yaml
+
+CMD ["/bin/provider-service"]
diff --git a/provider-service/Makefile b/provider-service/Makefile
@@ -0,0 +1,14 @@
+include ../docker-targets.mk
+
+IMG := kfp-operator-provider-service
+
+build:
+	CGO_ENABLED=0 GOOS=$(GOOS) GOARCH=$(GOARCH) go build -o bin/provider-service ./main.go
+
+test:
+	go test ./... -tags=unit
+
+#functional-test: docker-build
+#	docker compose up -d
+#	go test ./... -tags=functional
+#	docker compose down