fix trigger joins (#340)

* fix trigger joins

* remove unused imports

operator/pkg/cli/kafka.go

@@ -252,7 +252,8 @@ func showKafkaOutputMsg(e *kafka.Message, tensor string) error {
 	res := &v2_dataplane.ModelInferResponse{}
 	err := proto.Unmarshal(e.Value, res)
 	if err != nil {
-		return err
+		fmt.Printf("%s:%s\n", string(e.Key), string(e.Value))
+		return nil
 	}
 	err = updateResponseFromRawContents(res)
 	if err != nil {
@@ -261,12 +262,12 @@ func showKafkaOutputMsg(e *kafka.Message, tensor string) error {
 	if tensor != "" {
 		for _, output := range res.Outputs {
 			if output.Name == tensor {
-				printProto(output)
+				printProtoWithKey(e.Key, output)
 			}
 		}
 
 	} else {
-		printProto(res)
+		printProtoWithKey(e.Key, res)
 	}
 	return nil
 }
@@ -275,7 +276,8 @@ func showKafkaInputMsg(e *kafka.Message, tensor string) error {
 	req := &v2_dataplane.ModelInferRequest{}
 	err := proto.Unmarshal(e.Value, req)
 	if err != nil {
-		return err
+		fmt.Printf("%s:%s\n", string(e.Key), string(e.Value))
+		return nil
 	}
 	err = updateRequestFromRawContents(req)
 	if err != nil {
@@ -284,12 +286,12 @@ func showKafkaInputMsg(e *kafka.Message, tensor string) error {
 	if tensor != "" {
 		for _, input := range req.Inputs {
 			if input.Name == tensor {
-				printProto(input)
+				printProtoWithKey(e.Key, input)
 			}
 		}
 
 	} else {
-		printProto(req)
+		printProtoWithKey(e.Key, req)
 	}
 	return nil
 }

operator/pkg/cli/scheduler.go

@@ -66,6 +66,15 @@ func printProto(msg proto.Message) {
 	}
 }
 
+func printProtoWithKey(key []byte, msg proto.Message) {
+	resJson, err := protojson.Marshal(msg)
+	if err != nil {
+		fmt.Printf("Failed to print proto: %s", err.Error())
+	} else {
+		fmt.Printf("%s:%s\n", string(key), string(resJson))
+	}
+}
+
 func unMarshallYamlStrict(data []byte, msg interface{}) error {
 	jsonData, err := yaml.YAMLToJSON(data)
 	if err != nil {

samples/models/id1_node.yaml

@@ -0,0 +1,10 @@
+apiVersion: mlops.seldon.io/v1alpha1
+kind: Model
+metadata:
+  name: id1_node
+  namespace: seldon-mesh
+spec:
+  storageUri: "gs://seldon-models/triton/id_node"
+  requirements:
+  - triton
+  - python

samples/models/id2_node.yaml

@@ -0,0 +1,10 @@
+apiVersion: mlops.seldon.io/v1alpha1
+kind: Model
+metadata:
+  name: id2_node
+  namespace: seldon-mesh
+spec:
+  storageUri: "gs://seldon-models/triton/id_node"
+  requirements:
+  - triton
+  - python

samples/models/join_node.yaml

@@ -0,0 +1,10 @@
+apiVersion: mlops.seldon.io/v1alpha1
+kind: Model
+metadata:
+  name: join_node
+  namespace: seldon-mesh
+spec:
+  storageUri: "gs://seldon-models/triton/join_node"
+  requirements:
+  - triton
+  - python

samples/pipelines/triggers_join_inputs.yaml

@@ -0,0 +1,18 @@
+apiVersion: mlops.seldon.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: triggers_join_inputs
+  namespace: seldon-mesh
+spec:
+  steps:
+    - name: join_node
+      inputs:
+        - triggers_join_inputs.inputs.INPUT1
+        - triggers_join_inputs.inputs.INPUT2
+      triggers:
+        - triggers_join_inputs.inputs.TRIGGER1
+        - triggers_join_inputs.inputs.TRIGGER2
+      triggersJoinType: any
+  output:
+    steps:
+      - join_node

samples/pipelines/triggers_join_internal.yaml

@@ -0,0 +1,28 @@
+apiVersion: mlops.seldon.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: triggers_join_internal
+  namespace: seldon-mesh
+spec:
+  steps:
+    - name: id1_node
+      inputs:
+        - triggers_join_internal.inputs.TRIGGER1
+      tensorMap:
+        triggers_join_internal.inputs.TRIGGER1: INPUT1
+    - name: id2_node
+      inputs:
+        - triggers_join_internal.inputs.TRIGGER2
+      tensorMap:
+        triggers_join_internal.inputs.TRIGGER2: INPUT1
+    - name: join_node
+      inputs:
+        - triggers_join_internal.inputs.INPUT1
+        - triggers_join_internal.inputs.INPUT2
+      triggers:
+        - id1_node.outputs.OUTPUT1
+        - id2_node.outputs.OUTPUT1
+      triggersJoinType: any
+  output:
+    steps:
+      - join_node