Log-based events

apis/pipelines/v1alpha5/test_generators.go

@@ -83,22 +83,27 @@ func RandomRunSchedule() *RunSchedule {
 	}
 }
 
+func RandomOutputArtifact() OutputArtifact {
+	return OutputArtifact{
+		Name: RandomString(),
+		Path: ArtifactPath{
+			Locator: ArtifactLocator{
+				Component: RandomString(),
+				Artifact:  RandomString(),
+				Index:     rand.Int(),
+			},
+			Filter: RandomString(),
+		},
+	}
+}
+
 func RandomRunScheduleSpec() RunScheduleSpec {
 	return RunScheduleSpec{
 		Pipeline:          PipelineIdentifier{Name: RandomString(), Version: RandomString()},
 		ExperimentName:    RandomString(),
 		RuntimeParameters: RandomNamedValues(),
-		Artifacts: RandomList(func() OutputArtifact {
-			return OutputArtifact{Name: RandomString(), Path: ArtifactPath{
-				Locator: ArtifactLocator{
-					Component: RandomString(),
-					Artifact:  RandomString(),
-					Index:     rand.Int(),
-				},
-				Filter: RandomString(),
-			}}
-		}),
-		Schedule: RandomString(),
+		Artifacts:         RandomList(RandomOutputArtifact),
+		Schedule:          RandomString(),
 	}
 }
 

argo/providers/base/k8s_api.go

@@ -0,0 +1,103 @@
+package base
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/sky-uk/kfp-operator/apis/pipelines"
+	pipelinesv1 "github.com/sky-uk/kfp-operator/apis/pipelines/v1alpha5"
+	"github.com/sky-uk/kfp-operator/argo/common"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/dynamic"
+)
+
+var RunGVR = pipelinesv1.GroupVersion.WithResource("runs")
+var RunConfigurationGVR = pipelinesv1.GroupVersion.WithResource("runconfigurations")
+
+func artifactNamePathAsString(unstructuredArtifact map[string]interface{}) (string, string, error) {
+	nameStr, ok := unstructuredArtifact["name"].(string)
+	if !ok {
+		err := errors.New("name in artifact is malformed")
+		return "", "", err
+	}
+	pathStr, ok := unstructuredArtifact["path"].(string)
+	if !ok {
+		err := errors.New("path in artifact is malformed")
+		return "", "", err
+	}
+
+	return nameStr, pathStr, nil
+}
+
+func artifactsForFields(ctx context.Context, obj *unstructured.Unstructured, fields ...string) ([]pipelinesv1.OutputArtifact, error) {
+	logger := common.LoggerFromContext(ctx)
+	artifactsField, hasArtifacts, err := unstructured.NestedSlice(obj.Object, fields...)
+	if err != nil || !hasArtifacts {
+		logger.Error(err, "Failed to get artifacts out of unstructured")
+		return nil, err
+	}
+
+	return pipelines.MapErr(artifactsField, func(fa interface{}) (pipelinesv1.OutputArtifact, error) {
+		unstructuredArtifact, ok := fa.(map[string]interface{})
+		if !ok {
+			err = errors.New("artifacts malformed")
+			logger.Error(err, "Failed to cast")
+			return pipelinesv1.OutputArtifact{}, err
+		}
+
+		nameStr, pathStr, err := artifactNamePathAsString(unstructuredArtifact)
+		if err != nil {
+			logger.Error(err, "Failed to extract data from unstructured artifact")
+			return pipelinesv1.OutputArtifact{}, err
+		}
+		path, err := pipelinesv1.ArtifactPathFromString(pathStr)
+		if err != nil {
+			logger.Error(err, fmt.Sprintf("Failed to process path %+v", pathStr))
+			return pipelinesv1.OutputArtifact{}, err
+		}
+
+		return pipelinesv1.OutputArtifact{
+			Name: nameStr,
+			Path: path,
+		}, nil
+	})
+}
+
+func artifactsForUnstructured(ctx context.Context, obj *unstructured.Unstructured, gvr schema.GroupVersionResource) ([]pipelinesv1.OutputArtifact, error) {
+	if gvr == RunGVR {
+		return artifactsForFields(ctx, obj, "spec", "artifacts")
+	}
+
+	if gvr == RunConfigurationGVR {
+		return artifactsForFields(ctx, obj, "spec", "run", "artifacts")
+	}
+
+	return nil, errors.New("unhandled resource, only runs and runconfigurations expected")
+}
+
+func CreateK8sClient() (dynamic.Interface, error) {
+	k8sConfig, err := common.K8sClientConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	return dynamic.NewForConfig(k8sConfig)
+}
+
+type K8sApi struct {
+	K8sClient dynamic.Interface
+}
+
+func (k8a K8sApi) GetRunArtifactDefinitions(ctx context.Context, namespacedName types.NamespacedName, gvr schema.GroupVersionResource) ([]pipelinesv1.OutputArtifact, error) {
+	obj, err := k8a.K8sClient.Resource(gvr).Namespace(namespacedName.Namespace).Get(ctx, namespacedName.Name, metav1.GetOptions{})
+
+	if err != nil {
+		common.LoggerFromContext(ctx).Error(err, fmt.Sprintf("Failed to retrieve resource %+v", gvr))
+		return nil, err
+	}
+
+	return artifactsForUnstructured(ctx, obj, gvr)
+}

argo/providers/base/k8s_api_test.go

@@ -0,0 +1,155 @@
+//go:build unit
+
+package base
+
+import (
+	"context"
+	"github.com/go-logr/logr"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/sky-uk/kfp-operator/apis"
+	"github.com/sky-uk/kfp-operator/apis/pipelines/v1alpha5"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+var _ = Context("K8sApi", func() {
+	ctx := logr.NewContext(context.Background(), logr.Discard())
+
+	Describe("artifactsForUnstructured", func() {
+		unstructuredArtifacts := make([]interface{}, 2)
+		name1 := "somename"
+		name2 := "someothername"
+		path1Str := "Pusher:pushed_model:0[pushed == 1]"
+		path2Str := "Pusher:pushed_model:1[pushed == 2]"
+
+		unstructuredArtifacts[0] = map[string]interface{}{
+			"name": name1,
+			"path": path1Str,
+		}
+		unstructuredArtifacts[1] = map[string]interface{}{
+			"name": name2,
+			"path": path2Str,
+		}
+
+		expectedArtifacts := make([]v1alpha5.OutputArtifact, 2)
+		path1, err := v1alpha5.ArtifactPathFromString(path1Str)
+		Expect(err).NotTo(HaveOccurred())
+		path2, err := v1alpha5.ArtifactPathFromString(path2Str)
+		Expect(err).NotTo(HaveOccurred())
+		expectedArtifacts[0] = v1alpha5.OutputArtifact{
+			Name: name1,
+			Path: path1,
+		}
+		expectedArtifacts[1] = v1alpha5.OutputArtifact{
+			Name: name2,
+			Path: path2,
+		}
+
+		When("a run has artifacts", func() {
+			It("returns the artifacts", func() {
+				obj := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"spec": map[string]interface{}{
+							"artifacts": unstructuredArtifacts,
+						},
+					},
+				}
+
+				artifacts, err := artifactsForUnstructured(ctx, obj, RunGVR)
+
+				Expect(err).NotTo(HaveOccurred())
+				Expect(artifacts).To(Equal(expectedArtifacts))
+			})
+		})
+
+		When("a run has no artifacts", func() {
+			It("returns no artifacts", func() {
+				obj := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"spec": map[string]interface{}{},
+					},
+				}
+
+				artifacts, err := artifactsForUnstructured(ctx, obj, RunGVR)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(artifacts).To(BeEmpty())
+			})
+		})
+
+		When("a run is malformed", func() {
+			It("errors", func() {
+				obj := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"spec": map[string]interface{}{
+							"artifacts": apis.RandomString(),
+						},
+					},
+				}
+
+				_, err := artifactsForUnstructured(ctx, obj, RunGVR)
+				Expect(err).To(HaveOccurred())
+			})
+		})
+
+		When("a run configuration has artifacts", func() {
+			It("returns the artifacts", func() {
+				obj := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"spec": map[string]interface{}{
+							"run": map[string]interface{}{
+								"artifacts": unstructuredArtifacts,
+							},
+						},
+					},
+				}
+
+				artifacts, err := artifactsForUnstructured(ctx, obj, RunConfigurationGVR)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(artifacts).To(Equal(expectedArtifacts))
+			})
+		})
+
+		When("a run configuration has no artifacts", func() {
+			It("returns no artifacts", func() {
+				obj := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"run": map[string]interface{}{},
+					},
+				}
+
+				artifacts, err := artifactsForUnstructured(ctx, obj, RunConfigurationGVR)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(artifacts).To(BeEmpty())
+			})
+		})
+
+		When("a run configuration is malformed", func() {
+			It("errors", func() {
+				obj := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"spec": map[string]interface{}{
+							"run": map[string]interface{}{
+								"artifacts": apis.RandomString(),
+							},
+						},
+					},
+				}
+
+				_, err := artifactsForUnstructured(ctx, obj, RunConfigurationGVR)
+				Expect(err).To(HaveOccurred())
+			})
+		})
+
+		When("an unknown resource is provided", func() {
+			It("errors", func() {
+				obj := &unstructured.Unstructured{
+					Object: map[string]interface{}{},
+				}
+
+				_, err := artifactsForUnstructured(ctx, obj, schema.GroupVersionResource{Group: apis.RandomString(), Version: apis.RandomString(), Resource: apis.RandomString()})
+				Expect(err).To(HaveOccurred())
+			})
+		})
+	})
+})

argo/providers/kfp/provider.go

@@ -19,7 +19,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 	"gopkg.in/yaml.v2"
-	"k8s.io/client-go/dynamic"
 	"os"
 )
 
@@ -341,15 +340,6 @@ func (kfpp KfpProvider) DeleteExperiment(ctx context.Context, providerConfig Kfp
 	return err
 }
 
-func createK8sClient() (dynamic.Interface, error) {
-	k8sConfig, err := common.K8sClientConfig()
-	if err != nil {
-		return nil, err
-	}
-
-	return dynamic.NewForConfig(k8sConfig)
-}
-
 func ConnectToMetadataStore(address string) (*GrpcMetadataStore, error) {
 	conn, err := grpc.Dial(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
@@ -374,7 +364,7 @@ func ConnectToKfpApi(address string) (*GrpcKfpApi, error) {
 }
 
 func (kfpp KfpProvider) EventingServer(ctx context.Context, providerConfig KfpProviderConfig) (generic.EventingServer, error) {
-	k8sClient, err := createK8sClient()
+	k8sClient, err := CreateK8sClient()
 	if err != nil {
 		return nil, err
 	}
@@ -391,9 +381,9 @@ func (kfpp KfpProvider) EventingServer(ctx context.Context, providerConfig KfpPr
 
 	return &KfpEventingServer{
 		ProviderConfig: providerConfig,
-		K8sClient:     k8sClient,
-		Logger:        common.LoggerFromContext(ctx),
-		MetadataStore: metadataStore,
-		KfpApi:        kfpApi,
+		K8sClient:      k8sClient,
+		Logger:         common.LoggerFromContext(ctx),
+		MetadataStore:  metadataStore,
+		KfpApi:         kfpApi,
 	}, nil
 }