Explore: Add transformations to correlation data links

.betterer.results

@@ -3574,9 +3574,6 @@ exports[`better eslint`] = {
       [0, 0, 0, "Do not use any type assertions.", "4"],
       [0, 0, 0, "Do not use any type assertions.", "5"]
     ],
-    "public/app/features/explore/utils/links.ts:5381": [
-      [0, 0, 0, "Unexpected any. Specify a different type.", "0"]
-    ],
     "public/app/features/expressions/ExpressionDatasource.ts:5381": [
       [0, 0, 0, "Unexpected any. Specify a different type.", "0"],
       [0, 0, 0, "Do not use any type assertions.", "1"],

package.json

@@ -140,7 +140,7 @@
     "@types/js-yaml": "^4.0.5",
     "@types/jsurl": "^1.2.28",
     "@types/lodash": "4.14.187",
-    "@types/logfmt": "^1.2.1",
+    "@types/logfmt": "^1.2.3",
     "@types/mousetrap": "1.6.10",
     "@types/node": "18.14.0",
     "@types/ol-ext": "npm:@siedlerchr/types-ol-ext@3.0.6",

packages/grafana-data/src/types/ScopedVars.ts

@@ -1,5 +1,5 @@
 export interface ScopedVar<T = any> {
-  text: any;
+  text?: any;
   value: T;
   [key: string]: any;
 }

packages/grafana-data/src/types/dataLink.ts

@@ -40,12 +40,27 @@ export interface DataLink<T extends DataQuery = any> {
   internal?: InternalDataLink<T>;
 }
 
+/** @internal */
+export enum SupportedTransformationTypes {
+  Regex = 'regex',
+  Logfmt = 'logfmt',
+}
+
+/** @internal */
+export interface DataLinkTransformationConfig {
+  type: SupportedTransformationTypes;
+  field?: string;
+  expression?: string;
+  mapValue?: string;
+}
+
 /** @internal */
 export interface InternalDataLink<T extends DataQuery = any> {
   query: T;
   datasourceUid: string;
   datasourceName: string; // used as a title if `DataLink.title` is empty
   panelsState?: ExplorePanelsState;
+  transformations?: DataLinkTransformationConfig[];
 }
 
 export type LinkTarget = '_blank' | '_self' | undefined;

pkg/services/correlations/database.go

@@ -129,6 +129,9 @@ func (s CorrelationsService) updateCorrelation(ctx context.Context, cmd UpdateCo
 			if cmd.Config.Target != nil {
 				correlation.Config.Target = *cmd.Config.Target
 			}
+			if cmd.Config.Transformations != nil {
+				correlation.Config.Transformations = cmd.Config.Transformations
+			}
 		}
 
 		updateCount, err := session.Where("uid = ? AND source_uid = ?", correlation.UID, correlation.SourceUID).Limit(1).Update(correlation)

pkg/services/correlations/models.go

@@ -14,10 +14,21 @@ var (
 	ErrCorrelationNotFound                = errors.New("correlation not found")
 	ErrUpdateCorrelationEmptyParams       = errors.New("not enough parameters to edit correlation")
 	ErrInvalidConfigType                  = errors.New("invalid correlation config type")
+	ErrInvalidTransformationType          = errors.New("invalid transformation type")
+	ErrTransformationNotNested            = errors.New("transformations must be nested under config")
+	ErrTransformationRegexReqExp          = errors.New("regex transformations require expression")
 )
 
 type CorrelationConfigType string
 
+type Transformation struct {
+	//Enum: regex,logfmt
+	Type       string `json:"type"`
+	Expression string `json:"expression,omitempty"`
+	Field      string `json:"field,omitempty"`
+	MapValue   string `json:"mapValue,omitempty"`
+}
+
 const (
 	ConfigTypeQuery CorrelationConfigType = "query"
 )
@@ -29,6 +40,19 @@ func (t CorrelationConfigType) Validate() error {
 	return nil
 }
 
+func (t Transformations) Validate() error {
+	for _, v := range t {
+		if v.Type != "regex" && v.Type != "logfmt" {
+			return fmt.Errorf("%s: \"%s\"", ErrInvalidTransformationType, t)
+		} else if v.Type == "regex" && len(v.Expression) == 0 {
+			return fmt.Errorf("%s: \"%s\"", ErrTransformationRegexReqExp, t)
+		}
+	}
+	return nil
+}
+
+type Transformations []Transformation
+
 // swagger:model
 type CorrelationConfig struct {
 	// Field used to attach the correlation link
@@ -42,21 +66,28 @@ type CorrelationConfig struct {
 	// required:true
 	// example: { "expr": "job=app" }
 	Target map[string]interface{} `json:"target" binding:"Required"`
+	// Source data transformations
+	// required:false
+	// example: [{"type": "logfmt"}]
+	Transformations Transformations `json:"transformations,omitempty"`
 }
 
 func (c CorrelationConfig) MarshalJSON() ([]byte, error) {
 	target := c.Target
+	transformations := c.Transformations
 	if target == nil {
 		target = map[string]interface{}{}
 	}
 	return json.Marshal(struct {
-		Type   CorrelationConfigType  `json:"type"`
-		Field  string                 `json:"field"`
-		Target map[string]interface{} `json:"target"`
+		Type            CorrelationConfigType  `json:"type"`
+		Field           string                 `json:"field"`
+		Target          map[string]interface{} `json:"target"`
+		Transformations Transformations        `json:"transformations,omitempty"`
 	}{
-		Type:   ConfigTypeQuery,
-		Field:  c.Field,
-		Target: target,
+		Type:            ConfigTypeQuery,
+		Field:           c.Field,
+		Target:          target,
+		Transformations: transformations,
 	})
 }
 
@@ -117,6 +148,10 @@ func (c CreateCorrelationCommand) Validate() error {
 	if c.TargetUID == nil && c.Config.Type == ConfigTypeQuery {
 		return fmt.Errorf("correlations of type \"%s\" must have a targetUID", ConfigTypeQuery)
 	}
+
+	if err := c.Config.Transformations.Validate(); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -151,6 +186,9 @@ type CorrelationConfigUpdateDTO struct {
 	// Target data query
 	// example: { "expr": "job=app" }
 	Target *map[string]interface{} `json:"target"`
+	// Source data transformations
+	// example: [{"type": "logfmt"},{"type":"regex","expression":"(Superman|Batman)", "variable":"name"}]
+	Transformations []Transformation `json:"transformations"`
 }
 
 func (c CorrelationConfigUpdateDTO) Validate() error {

pkg/services/provisioning/datasources/datasources.go

@@ -152,6 +152,10 @@ func makeCreateCorrelationCommand(correlation map[string]interface{}, SourceUID
 		createCommand.TargetUID = &targetUID
 	}
 
+	if correlation["transformations"] != nil {
+		return correlations.CreateCorrelationCommand{}, correlations.ErrTransformationNotNested
+	}
+
 	if correlation["config"] != nil {
 		jsonbody, err := json.Marshal(correlation["config"])
 		if err != nil {

pkg/tests/api/correlations/correlations_create_test.go

@@ -237,6 +237,8 @@ func TestIntegrationCreateCorrelation(t *testing.T) {
 		label := "a label"
 		fieldName := "fieldName"
 		configType := correlations.ConfigTypeQuery
+		transformation := correlations.Transformation{Type: "logfmt"}
+		transformation2 := correlations.Transformation{Type: "regex", Expression: "testExpression", MapValue: "testVar"}
 		res := ctx.Post(PostParams{
 			url: fmt.Sprintf("/api/datasources/uid/%s/correlations", writableDs),
 			body: fmt.Sprintf(`{
@@ -246,7 +248,11 @@ func TestIntegrationCreateCorrelation(t *testing.T) {
 					"config": {
 						"type": "%s",
 						"field": "%s",
-						"target": { "expr": "foo" }
+						"target": { "expr": "foo" },
+						"transformations": [
+							{"type": "logfmt"},
+							{"type": "regex", "expression": "testExpression", "mapValue": "testVar"}
+						]
 					}
 				}`, writableDs, description, label, configType, fieldName),
 			user: adminUser,
@@ -268,6 +274,8 @@ func TestIntegrationCreateCorrelation(t *testing.T) {
 		require.Equal(t, configType, response.Result.Config.Type)
 		require.Equal(t, fieldName, response.Result.Config.Field)
 		require.Equal(t, map[string]interface{}{"expr": "foo"}, response.Result.Config.Target)
+		require.Equal(t, transformation, response.Result.Config.Transformations[0])
+		require.Equal(t, transformation2, response.Result.Config.Transformations[1])
 
 		require.NoError(t, res.Body.Close())
 	})

pkg/tests/api/correlations/correlations_read_test.go

@@ -79,6 +79,9 @@ func TestIntegrationReadCorrelation(t *testing.T) {
 			Type:   correlations.ConfigTypeQuery,
 			Field:  "foo",
 			Target: map[string]interface{}{},
+			Transformations: []correlations.Transformation{
+				{Type: "logfmt"},
+			},
 		},
 	})
 

pkg/tests/api/correlations/correlations_update_test.go

@@ -287,7 +287,8 @@ func TestIntegrationUpdateCorrelation(t *testing.T) {
 				"config": {
 					"field": "field",
 					"type": "query",
-					"target": { "expr": "bar" }
+					"target": { "expr": "bar" },
+					"transformations": [ {"type": "logfmt"} ]
 				}
 			}`,
 		})
@@ -305,6 +306,7 @@ func TestIntegrationUpdateCorrelation(t *testing.T) {
 		require.Equal(t, "1", response.Result.Description)
 		require.Equal(t, "field", response.Result.Config.Field)
 		require.Equal(t, map[string]interface{}{"expr": "bar"}, response.Result.Config.Target)
+		require.Equal(t, correlations.Transformation{Type: "logfmt"}, response.Result.Config.Transformations[0])
 		require.NoError(t, res.Body.Close())
 
 		// partially updating only label

public/app/core/utils/explore.ts

@@ -192,7 +192,7 @@ export const safeParseJson = (text?: string): any | undefined => {
 };
 
 export const safeStringifyValue = (value: any, space?: number) => {
-  if (!value) {
+  if (value === undefined || value === null) {
     return '';
   }
 

public/app/features/correlations/transformations.ts

@@ -0,0 +1,34 @@
+import logfmt from 'logfmt';
+
+import { ScopedVars, DataLinkTransformationConfig, SupportedTransformationTypes } from '@grafana/data';
+import { safeStringifyValue } from 'app/core/utils/explore';
+
+export const getTransformationVars = (
+  transformation: DataLinkTransformationConfig,
+  fieldValue: string,
+  fieldName: string
+): ScopedVars => {
+  let transformationScopedVars: ScopedVars = {};
+  let transformVal: { [key: string]: string | boolean | null | undefined } = {};
+  if (transformation.type === SupportedTransformationTypes.Regex && transformation.expression) {
+    const regexp = new RegExp(transformation.expression, 'gi');
+    const matches = fieldValue.matchAll(regexp);
+    for (const match of matches) {
+      if (match.groups) {
+        transformVal = match.groups;
+      } else {
+        transformVal[transformation.mapValue || fieldName] = match[1] || match[0];
+      }
+    }
+  } else if (transformation.type === SupportedTransformationTypes.Logfmt) {
+    transformVal = logfmt.parse(fieldValue);
+  }
+
+  Object.keys(transformVal).forEach((key) => {
+    const transformValString =
+      typeof transformVal[key] === 'string' ? transformVal[key] : safeStringifyValue(transformVal[key]);
+    transformationScopedVars[key] = { value: transformValString };
+  });
+
+  return transformationScopedVars;
+};

public/app/features/correlations/types.ts

@@ -1,14 +1,18 @@
+import { DataLinkTransformationConfig } from '@grafana/data';
+
 export interface AddCorrelationResponse {
   correlation: Correlation;
 }
 
 export type GetCorrelationsResponse = Correlation[];
 
 type CorrelationConfigType = 'query';
+
 export interface CorrelationConfig {
   field: string;
   target: object;
   type: CorrelationConfigType;
+  transformations?: DataLinkTransformationConfig[];
 }
 
 export interface Correlation {

public/app/features/correlations/utils.ts

@@ -39,6 +39,7 @@ const decorateDataFrameWithInternalDataLinks = (dataFrame: DataFrame, correlatio
             query: correlation.config?.target,
             datasourceUid: correlation.target.uid,
             datasourceName: correlation.target.name,
+            transformations: correlation.config?.transformations,
           },
           url: '',
           title: correlation.label || correlation.target.name,