feat: Meta-data. Fixes #161 (#314)

README.md

@@ -69,6 +69,7 @@ Intermediate:
 * [Scaling](docs/SCALING.md)
 * [Command line](docs/CLI.md)
 * [Expression syntax](docs/EXPRESSIONS.md)
+* [Meta-data](docs/META.md)
 * [Workflow interop](docs/WORKFLOW_INTEROP.md)
 * [Events interop](docs/EVENTS_INTEROP.md)
 * [Kubectl](docs/KUBECTL.md)

api/v1alpha1/abstract_volume_source.go

@@ -0,0 +1,26 @@
+package v1alpha1
+
+import (
+	"fmt"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
+type AbstractVolumeSource corev1.VolumeSource
+
+func (in AbstractVolumeSource) getURNParts() (kind string, name string) {
+	if v := in.ConfigMap; v != nil {
+		return "configmap", v.Name
+	} else if v := in.PersistentVolumeClaim; v != nil {
+		return "persistentvolumeclaim", v.ClaimName
+	} else if v := in.Secret; v != nil {
+		return "secret", v.SecretName
+	}
+	panic(fmt.Errorf("un-suppported volume source %v", in))
+}
+
+func (in AbstractVolumeSource) GenURN(cluster, namespace string) string {
+	kind, name := in.getURNParts()
+	return fmt.Sprintf("urn:dataflow:volume:%s:%s.%s.%s.%s", strings.ToLower(kind), name, kind, namespace, cluster)
+}

api/v1alpha1/abstract_volume_source_test.go

@@ -0,0 +1,39 @@
+package v1alpha1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+)
+
+var cluster, namespace = "my-cluster", "my-ns"
+
+func TestAbstractVolumeSource_GenURN(t *testing.T) {
+	t.Run("ConfigMap", func(t *testing.T) {
+		urn := AbstractVolumeSource{
+			ConfigMap: &corev1.ConfigMapVolumeSource{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: "my-name",
+				},
+			},
+		}.GenURN(cluster, namespace)
+		assert.Equal(t, "urn:dataflow:volume:configmap:my-name.configmap.my-ns.my-cluster", urn)
+	})
+	t.Run("PersistentVolumeClaim", func(t *testing.T) {
+		urn := AbstractVolumeSource{
+			PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{
+				ClaimName: "my-name",
+			},
+		}.GenURN(cluster, namespace)
+		assert.Equal(t, "urn:dataflow:volume:persistentvolumeclaim:my-name.persistentvolumeclaim.my-ns.my-cluster", urn)
+	})
+	t.Run("Secret", func(t *testing.T) {
+		urn := AbstractVolumeSource{
+			Secret: &corev1.SecretVolumeSource{
+				SecretName: "my-name",
+			},
+		}.GenURN(cluster, namespace)
+		assert.Equal(t, "urn:dataflow:volume:secret:my-name.secret.my-ns.my-cluster", urn)
+	})
+}

api/v1alpha1/const.go

@@ -23,6 +23,7 @@ const (
 	EnvImagePrefix      = "ARGO_DATAFLOW_IMAGE_PREFIX" // default "quay.io/argoproj"
 	EnvNamespace        = "ARGO_DATAFLOW_NAMESPACE"
 	EnvPipelineName     = "ARGO_DATAFLOW_PIPELINE_NAME"
+	EnvPod              = "ARGO_DATAFLOW_POD"
 	EnvReplica          = "ARGO_DATAFLOW_REPLICA"
 	EnvStep             = "ARGO_DATAFLOW_STEP"
 	EnvPeekDelay        = "ARGO_DATAFLOW_PEEK_DELAY"         // how long between peeking (default 4m)

api/v1alpha1/cron.go

@@ -1,7 +1,15 @@
 package v1alpha1
 
+import (
+	"fmt"
+)
+
 type Cron struct {
 	Schedule string `json:"schedule" protobuf:"bytes,1,opt,name=schedule"`
 	// +kubebuilder:default="2006-01-02T15:04:05Z07:00"
 	Layout string `json:"layout,omitempty" protobuf:"bytes,2,opt,name=layout"`
 }
+
+func (in Cron) GenURN(cluster, namespace string) string {
+	return fmt.Sprintf("urn:dataflow:cron:%s", in.Schedule)
+}

api/v1alpha1/cron_test.go

@@ -0,0 +1,12 @@
+package v1alpha1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCron_GenURN(t *testing.T) {
+	urn := Cron{Schedule: "* * * * *"}.GenURN(cluster, namespace)
+	assert.Equal(t, "urn:dataflow:cron:* * * * *", urn)
+}

api/v1alpha1/db.go

@@ -1,13 +1,25 @@
 package v1alpha1
 
-import corev1 "k8s.io/api/core/v1"
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+)
 
 type Database struct {
 	// +kubebuilder:default=default
 	Driver     string        `json:"driver,omitempty" protobuf:"bytes,1,opt,name=driver"`
 	DataSource *DBDataSource `json:"dataSource,omitempty" protobuf:"bytes,2,opt,name=dataSource"`
 }
 
+func (in Database) GenURN(cluster, namespace string) string {
+	if in.DataSource.Value != "" {
+		return fmt.Sprintf("urn:dataflow:db:%s", in.DataSource.Value)
+	} else {
+		return fmt.Sprintf("urn:dataflow:db:%s.secret.%s.%s:%s", in.DataSource.ValueFrom.SecretKeyRef.Name, namespace, cluster, in.DataSource.ValueFrom.SecretKeyRef.Key)
+	}
+}
+
 type DBDataSource struct {
 	Value     string            `json:"value,omitempty" protobuf:"bytes,1,opt,name=value"`
 	ValueFrom *DBDataSourceFrom `json:"valueFrom,omitempty" protobuf:"bytes,2,opt,name=valueFrom"`

api/v1alpha1/db_test.go

@@ -0,0 +1,22 @@
+package v1alpha1
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func TestDatabase_GenURN(t *testing.T) {
+	t.Run("Value", func(t *testing.T) {
+		urn := Database{DataSource: &DBDataSource{Value: "my-value"}}.GenURN(cluster, namespace)
+		assert.Equal(t, "urn:dataflow:db:my-value", urn)
+	})
+	t.Run("ValueFrom", func(t *testing.T) {
+		urn := Database{DataSource: &DBDataSource{ValueFrom: &DBDataSourceFrom{SecretKeyRef: &corev1.SecretKeySelector{
+			LocalObjectReference: corev1.LocalObjectReference{Name: "my-name"},
+			Key:                  "my-key",
+		}}}}.GenURN(cluster, namespace)
+		assert.Equal(t, "urn:dataflow:db:my-name.secret.my-ns.my-cluster:my-key", urn)
+	})
+}