Fix bson.D conversion in GRPC

pkg/storage/grpc_test.go

@@ -66,4 +66,54 @@ func TestRoundTripDataOverGRPC(t *testing.T) {
 	var gotThing1 bson.M
 	require.NoError(t, bson.Unmarshal(results[0], &gotThing1))
 	assert.Equal(t, thing1, gotThing1)
+
+	opts := plugins.EnsureIndexOptions{
+		Indices: []plugins.Index{
+			// query most recent outputs by run (porter installation run show, when we list outputs)
+			{Collection: CollectionOutputs, Keys: bson.D{{"namespace", 1}, {"installation", 1}, {"-resultId", 1}}},
+			// query outputs by result (list)
+			{Collection: CollectionOutputs, Keys: bson.D{{"resultId", 1}, {"name", 1}}, Unique: true},
+			// query most recent outputs by name for an installation
+			{Collection: CollectionOutputs, Keys: bson.D{{"namespace", 1}, {"installation", 1}, {"name", 1}, {"-resultId", 1}}},
+		},
+	}
+
+	err = client.EnsureIndex(ctx, opts)
+	require.NoError(t, err)
+
+	err = client.Insert(ctx, plugins.InsertOptions{
+		Collection: CollectionOutputs,
+		Documents: []bson.M{{"namespace": "dev", "installation": "test", "name": "thing1", "resultId": "111"},
+			{"namespace": "dev", "installation": "test", "name": "thing2", "resultId": "111"},
+			{"namespace": "dev", "installation": "test", "name": "thing2", "resultId": "222"},
+		},
+	})
+	require.NoError(t, err)
+
+	aggregateResults, err := client.Aggregate(ctx, plugins.AggregateOptions{
+		Collection: CollectionOutputs,
+		Pipeline: []bson.D{
+			// List outputs by installation
+			{{"$match", bson.M{
+				"namespace":    "dev",
+				"installation": "test",
+			}}},
+			// Reverse sort them (newest on top)
+			{{"$sort", bson.D{
+				{"namespace", 1},
+				{"installation", 1},
+				{"name", 1},
+				{"resultId", -1},
+			}}},
+			// Group them by output name and select the last value for each output
+			{{"$group", bson.D{
+				{"_id", "$name"},
+				{"lastOutput", bson.M{"$first": "$$ROOT"}},
+			}}},
+		},
+	})
+	require.NoError(t, err)
+	require.Len(t, aggregateResults, 2)
+	// make sure the group function is picking the most recent output value with the same name
+	require.Contains(t, aggregateResults[0].Lookup("lastOutput").String(), "222")
 }

pkg/storage/pluginstore/grpc.go

@@ -284,6 +284,34 @@ func ConvertBsonToPrimitives(src interface{}) interface{} {
 	}
 }
 
+// ConvertSliceToBsonD converts go slices to bson primitive.
+// it also works around a weirdness in how numbers are represented
+// by structpb.Value, where integer values are stored in float64. When we
+// deserialize from protobuf, this walks the specified value, finds ints
+// that were encoded as floats, and converts them back to ints.
+func ConvertSliceToBsonD(src interface{}) interface{} {
+	switch tv := src.(type) {
+	case []interface{}:
+		toBson := make(bson.D, 0, len(tv))
+		for _, item := range tv {
+			converted := ConvertSliceToBsonD(item)
+			if m, ok := converted.(map[string]interface{}); ok {
+				for k, v := range m {
+					toBson = append(toBson, bson.E{Key: k, Value: v})
+				}
+			}
+		}
+		return toBson
+	case map[string]interface{}:
+		for k, v := range tv {
+			tv[k] = ConvertSliceToBsonD(v)
+		}
+		return tv
+	default:
+		return tv
+	}
+}
+
 // ConvertFloatToInt works around a weirdness in how numbers are represented
 // by structpb.Value, where integer values are stored in float64. When we
 // deserialize from protobuf, this walks the specified value, finds ints
@@ -358,10 +386,14 @@ func NewStruct(src map[string]interface{}) *structpb.Struct {
 }
 
 // AsMap converts a protobuf struct into its original representation, bson.M.
-func AsMap(src *structpb.Struct) bson.M {
+func AsMap(src *structpb.Struct, c ...converter) bson.M {
 	dest := src.AsMap()
 	for k, v := range dest {
-		dest[k] = ConvertFloatToInt(v)
+		converted := ConvertFloatToInt(v)
+		for _, convert := range c {
+			converted = convert(v)
+		}
+		dest[k] = converted
 	}
 	return dest
 }
@@ -374,12 +406,14 @@ func AsMapList(src []*structpb.Struct) []bson.M {
 	return dest
 }
 
+type converter func(src interface{}) interface{}
+
 // AsOrderedMap converts an array of protobuf structs into its original
 // representation, bson.D.
-func AsOrderedMap(src []*structpb.Struct) bson.D {
+func AsOrderedMap(src []*structpb.Struct, c ...converter) bson.D {
 	dest := make(bson.D, 0, len(src))
 	for _, item := range src {
-		for k, v := range AsMap(item) {
+		for k, v := range AsMap(item, c...) {
 			dest = append(dest, bson.E{Key: k, Value: v})
 		}
 	}
@@ -391,7 +425,7 @@ func AsOrderedMap(src []*structpb.Struct) bson.D {
 func AsOrderedMapList(src []*proto.Stage) []bson.D {
 	dest := make([]bson.D, len(src))
 	for i, item := range src {
-		dest[i] = AsOrderedMap(item.Steps)
+		dest[i] = AsOrderedMap(item.Steps, ConvertSliceToBsonD)
 	}
 	return dest
 }

pkg/storage/pluginstore/grpc_test.go

@@ -71,11 +71,11 @@ func TestConvertBsonD(t *testing.T) {
 	}
 
 	tmp := FromOrderedMap(src)
-	dest := AsOrderedMap(tmp)
+	dest := AsOrderedMap(tmp, ConvertSliceToBsonD)
 
 	wantDest := bson.D{
 		{"a", "1"},
-		{"b", []interface{}{map[string]interface{}{"c": int64(1)}}},
+		{"b", bson.D{{"c", int64(1)}}},
 	}
 	require.Equal(t, wantDest, dest)
 }

tests/smoke/hello_test.go

@@ -122,4 +122,7 @@ func TestHelloBundle(t *testing.T) {
 	require.Error(t, err, "the chaos monkey should have failed the installation")
 	myLogs, _ := test.RequirePorter("installation", "outputs", "show", "mylogs", "-i=fail-with-outputs")
 	require.Contains(t, myLogs, "Hello, porterci")
+
+	myLogsListed, _ := test.RequirePorter("installation", "outputs", "list", "-i=fail-with-outputs")
+	require.Contains(t, myLogsListed, "Hello, porterci")
 }