Fix/pipelines temp work around for supporting dots in resource keys (#4064)

* chore: logs pipelines: add test for validating workaround for working with dots in keys

* fix: temp workaround for supporting pipeline filters using names with dots converted to underscore

pkg/query-service/app/logparsingpipeline/pipelineBuilder_test.go

@@ -482,3 +482,76 @@ func TestPipelineFilterWithStringOpsShouldNotSpamWarningsIfAttributeIsMissing(t
 		require.Equal(1, len(result))
 	}
 }
+
+func TestTemporaryWorkaroundForSupportingAttribsContainingDots(t *testing.T) {
+	// TODO(Raj): Remove this after dots are supported
+
+	require := require.New(t)
+
+	testPipeline := Pipeline{
+		OrderId: 1,
+		Name:    "pipeline1",
+		Alias:   "pipeline1",
+		Enabled: true,
+		Filter: &v3.FilterSet{
+			Operator: "AND",
+			Items: []v3.FilterItem{
+				{
+					Key: v3.AttributeKey{
+						Key:      "k8s_deployment_name",
+						DataType: v3.AttributeKeyDataTypeString,
+						Type:     v3.AttributeKeyTypeResource,
+					},
+					Operator: "=",
+					Value:    "ingress",
+				},
+			},
+		},
+		Config: []PipelineOperator{
+			{
+				ID:      "add",
+				Type:    "add",
+				Enabled: true,
+				Name:    "add",
+				Field:   "attributes.test",
+				Value:   "test-value",
+			},
+		},
+	}
+
+	testLogs := []model.SignozLog{{
+		Timestamp:         uint64(time.Now().UnixNano()),
+		Body:              "test log",
+		Attributes_string: map[string]string{},
+		Resources_string: map[string]string{
+			"k8s_deployment_name": "ingress",
+		},
+		SeverityText:   entry.Info.String(),
+		SeverityNumber: uint8(entry.Info),
+		SpanID:         "",
+		TraceID:        "",
+	}, {
+		Timestamp:         uint64(time.Now().UnixNano()),
+		Body:              "test log",
+		Attributes_string: map[string]string{},
+		Resources_string: map[string]string{
+			"k8s.deployment.name": "ingress",
+		},
+		SeverityText:   entry.Info.String(),
+		SeverityNumber: uint8(entry.Info),
+		SpanID:         "",
+		TraceID:        "",
+	}}
+
+	result, collectorWarnAndErrorLogs, err := SimulatePipelinesProcessing(
+		context.Background(),
+		[]Pipeline{testPipeline},
+		testLogs,
+	)
+	require.Nil(err)
+	require.Equal(0, len(collectorWarnAndErrorLogs), strings.Join(collectorWarnAndErrorLogs, "\n"))
+	require.Equal(2, len(result))
+	for _, processedLog := range result {
+		require.Equal(processedLog.Attributes_string["test"], "test-value")
+	}
+}

pkg/query-service/queryBuilderToExpr/queryBuilderToExpr.go

@@ -30,9 +30,9 @@ var logOperatorsToExpr = map[v3.FilterOperator]string{
 
 func getName(v v3.AttributeKey) string {
 	if v.Type == v3.AttributeKeyTypeTag {
-		return "attributes?." + v.Key
+		return fmt.Sprintf(`attributes["%s"]`, v.Key)
 	} else if v.Type == v3.AttributeKeyTypeResource {
-		return "resource?." + v.Key
+		return fmt.Sprintf(`resource["%s"]`, v.Key)
 	}
 	return v.Key
 }
@@ -53,8 +53,19 @@ func Parse(filters *v3.FilterSet) (string, error) {
 			return "", fmt.Errorf("operator not supported")
 		}
 
-		name := getName(v.Key)
+		// TODO(Raj): Remove the use of dot replaced alternative when key
+		// contains underscore after dots are supported in keys
+		names := []string{getName(v.Key)}
+		if strings.Contains(v.Key.Key, "_") {
+			dotKey := v.Key
+			dotKey.Key = strings.Replace(v.Key.Key, "_", ".", -1)
+			names = append(names, getName(dotKey))
+		}
+
+		filterParts := []string{}
+		for _, name := range names {
 			var filter string
+
 			switch v.Operator {
 			// uncomment following lines when new version of expr is used
 			// case v3.FilterOperatorIn, v3.FilterOperatorNotIn:
@@ -71,6 +82,11 @@ func Parse(filters *v3.FilterSet) (string, error) {
 				}
 			}
 
+			filterParts = append(filterParts, filter)
+		}
+
+		filter := strings.Join(filterParts, " || ")
+
 		// check if the filter is a correct expression language
 		_, err := expr.Compile(filter)
 		if err != nil {