[DX-1355] Simpler subject name derivation logic + new dashboard panel (#1970)

Author: Tofel

framework/.changeset/v0.10.3.md

@@ -0,0 +1 @@
+- Adds Container Critical level Node logs to CL Node Errors dashboard

framework/components/dockercompose/.changeset/v0.1.3.md

@@ -0,0 +1,2 @@
+- Simpler algorithm for creation message subject
+- Support for optional subject prefix

framework/components/dockercompose/chip_ingress_set/protos.go

@@ -23,23 +23,19 @@ type protoFile struct {
 }
 
 type ProtoSchemaSet struct {
-	Owner      string   `toml:"owner"`
-	Repository string   `toml:"repository"`
-	Ref        string   `toml:"ref"`     // ref or tag or commit SHA
-	Folders    []string `toml:"folders"` // if not provided, all protos will be fetched, otherwise only protos in these folders will be fetched
+	Owner         string   `toml:"owner"`
+	Repository    string   `toml:"repository"`
+	Ref           string   `toml:"ref"`            // ref or tag or commit SHA
+	Folders       []string `toml:"folders"`        // if not provided, all protos will be fetched, otherwise only protos in these folders will be fetched
+	SubjectPrefix string   `toml:"subject_prefix"` // optional prefix for subjects
 }
 
 // SubjectNamingStrategyFn is a function that is used to determine the subject name for a given proto file in a given repo
-type SubjectNamingStrategyFn func(path, source string, repoConfig ProtoSchemaSet) (string, error)
+type SubjectNamingStrategyFn func(subjectPrefix string, protoFile protoFile, repoConfig ProtoSchemaSet) (string, error)
 
 // RepositoryToSubjectNamingStrategyFn is a map of repository names to SubjectNamingStrategyFn functions
 type RepositoryToSubjectNamingStrategyFn map[string]SubjectNamingStrategyFn
 
-// DefaultRepositoryToSubjectNamingStrategy is a map of repository names to SubjectNamingStrategyFn functions
-var DefaultRepositoryToSubjectNamingStrategy = RepositoryToSubjectNamingStrategyFn{
-	"smartcontractkit/chainlink-protos": ChainlinkProtosSubjectNamingStrategy,
-}
-
 func ValidateRepoConfiguration(repoConfig ProtoSchemaSet) error {
 	if repoConfig.Owner == "" {
 		return errors.New("owner is required")
@@ -56,7 +52,7 @@ func ValidateRepoConfiguration(repoConfig ProtoSchemaSet) error {
 }
 
 func DefaultRegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSchemaSets []ProtoSchemaSet, schemaRegistryURL string) error {
-	return RegisterAndFetchProtos(ctx, client, protoSchemaSets, schemaRegistryURL, DefaultRepositoryToSubjectNamingStrategy)
+	return RegisterAndFetchProtos(ctx, client, protoSchemaSets, schemaRegistryURL, map[string]SubjectNamingStrategyFn{})
 }
 
 func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSchemaSets []ProtoSchemaSet, schemaRegistryURL string, repoToSubjectNamingStrategy RepositoryToSubjectNamingStrategyFn) error {
@@ -69,10 +65,10 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 		}
 
 		protoMap := make(map[string]string)
-		subjectMap := make(map[string]string)
+		subjects := make(map[string]string)
 
-		for _, pf := range protos {
-			protoMap[pf.Path] = pf.Content
+		for _, proto := range protos {
+			protoMap[proto.Path] = proto.Content
 
 			var subjectStrategy SubjectNamingStrategyFn
 			if strategy, ok := repoToSubjectNamingStrategy[protoSchemaSet.Owner+"/"+protoSchemaSet.Repository]; ok {
@@ -81,14 +77,14 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 				subjectStrategy = DefaultSubjectNamingStrategy
 			}
 
-			subject, nameErr := subjectStrategy(pf.Path, pf.Content, protoSchemaSet)
+			subjectMessage, nameErr := subjectStrategy(protoSchemaSet.SubjectPrefix, proto, protoSchemaSet)
 			if nameErr != nil {
-				return errors.Wrapf(nameErr, "failed to extract message name from %s", pf.Path)
+				return errors.Wrapf(nameErr, "failed to extract message name from %s", proto.Path)
 			}
-			subjectMap[pf.Path] = subject
+			subjects[proto.Path] = subjectMessage
 		}
 
-		registerErr := registerAllWithTopologicalSortingByTrial(schemaRegistryURL, protoMap, subjectMap)
+		registerErr := registerAllWithTopologicalSortingByTrial(schemaRegistryURL, protoMap, subjects)
 		if registerErr != nil {
 			return errors.Wrapf(registerErr, "failed to register protos from %s/%s", protoSchemaSet.Owner, protoSchemaSet.Repository)
 		}
@@ -97,46 +93,39 @@ func RegisterAndFetchProtos(ctx context.Context, client *github.Client, protoSch
 	return nil
 }
 
-func DefaultSubjectNamingStrategy(path, source string, protoSchemaSet ProtoSchemaSet) (string, error) {
-	messageName, nameErr := extractTopLevelMessageNamesWithRegex(source)
-	if nameErr != nil {
-		return "", errors.Wrapf(nameErr, "failed to extract message name from %s", path)
+func DefaultSubjectNamingStrategy(subjectPrefix string, proto protoFile, protoSchemaSet ProtoSchemaSet) (string, error) {
+	packageName, packageErr := extractPackageNameWithRegex(proto.Content)
+	if packageErr != nil {
+		return "", errors.Wrapf(packageErr, "failed to extract package name from %s", proto.Path)
 	}
-	return protoSchemaSet.Repository + "." + messageName, nil
-}
 
-// TODO once we have single source of truth for the relationship between protos and subjects, we need to modify this function
-func ChainlinkProtosSubjectNamingStrategy(path, source string, protoSchemaSet ProtoSchemaSet) (string, error) {
-	messageName, nameErr := extractTopLevelMessageNamesWithRegex(source)
+	messageNames, nameErr := extractTopLevelMessageNamesWithRegex(proto.Content)
 	if nameErr != nil {
-		return "", errors.Wrapf(nameErr, "failed to extract message name from %s", path)
+		return "", errors.Wrapf(nameErr, "failed to extract message name from %s", proto.Path)
 	}
+	messageName := messageNames[0]
 
-	// this only covers BaseMessage
-	if strings.HasPrefix(path, "common") {
-		return "cre-pb." + messageName, nil
+	return subjectPrefix + packageName + "." + messageName, nil
+}
+
+// extractPackageNameWithRegex extracts the package name from a proto source file using regex.
+// It returns an error if no package name is found.
+func extractPackageNameWithRegex(protoSrc string) (string, error) {
+	matches := regexp.MustCompile(`(?m)^\s*package\s+([a-zA-Z0-9.]+)\s*;`).FindStringSubmatch(protoSrc)
+	if len(matches) < 2 {
+		return "", fmt.Errorf("no package name found in proto source")
 	}
 
-	// this covers all other protos we currently have in the chainlink-protos repo
-	subject := "cre-workflows."
-	pathSplit := strings.Split(path, "/")
-	if len(pathSplit) > 1 {
-		for _, part := range pathSplit {
-			matches := regexp.MustCompile(`v[0-9]+`).FindAllStringSubmatch(part, -1)
-			if len(matches) > 0 {
-				subject += matches[0][0]
-			}
-		}
-	} else {
-		return "", fmt.Errorf("no subject found for %s", path)
+	if matches[1] == "" {
+		return "", fmt.Errorf("empty package name found in proto source")
 	}
 
-	return subject + "." + messageName, nil
+	return matches[1], nil
 }
 
 // we use simple regex to extract top-level message names from a proto file
 // so that we don't need to parse the proto file with a parser (which would require a lot of dependencies)
-func extractTopLevelMessageNamesWithRegex(protoSrc string) (string, error) {
+func extractTopLevelMessageNamesWithRegex(protoSrc string) ([]string, error) {
 	matches := regexp.MustCompile(`(?m)^\s*message\s+(\w+)\s*{`).FindAllStringSubmatch(protoSrc, -1)
 	var names []string
 	for _, match := range matches {
@@ -146,11 +135,10 @@ func extractTopLevelMessageNamesWithRegex(protoSrc string) (string, error) {
 	}
 
 	if len(names) == 0 {
-		return "", fmt.Errorf("no message names found in %s", protoSrc)
+		return nil, fmt.Errorf("no message names found in %s", protoSrc)
 	}
 
-	// even though there could be more than 1 message in a single proto, we still need to register all of them under one subject
-	return names[0], nil
+	return names, nil
 }
 
 // Fetches .proto files from a GitHub repo optionally scoped to specific folders. It is recommended to use `*github.Client` with auth token to avoid rate limiting.
@@ -280,10 +268,12 @@ func registerAllWithTopologicalSortingByTrial(
 				continue
 			}
 
+			singleProtoFailures := []error{}
 			framework.L.Debug().Msgf("🔄 registering %s as %s", path, subject)
 			_, registerErr := registerSingleProto(schemaRegistryURL, subject, schema.Source, refs)
 			if registerErr != nil {
 				failures = append(failures, fmt.Sprintf("%s: %v", path, registerErr))
+				singleProtoFailures = append(singleProtoFailures, registerErr)
 				continue
 			}
 
@@ -296,6 +286,7 @@ func registerAllWithTopologicalSortingByTrial(
 			})
 
 			framework.L.Info().Msgf("✔ registered: %s as %s", path, subject)
+
 			progress = true
 		}
 

framework/examples/myproject/smoke_chip_ingress_test.go

@@ -48,10 +48,11 @@ func TestChipIngressSmoke(t *testing.T) {
 
 		err := chipingressset.DefaultRegisterAndFetchProtos(ctx, client, []chipingressset.ProtoSchemaSet{
 			{
-				Owner:      "smartcontractkit",
-				Repository: "chainlink-protos",
-				Ref:        "626c42d55bdcb36dffe0077fff58abba40acc3e5",
-				Folders:    []string{"workflows"},
+				Owner:         "smartcontractkit",
+				Repository:    "chainlink-protos",
+				Ref:           "95decc005a91a1fd2621af9d9f00cb36d8061067",
+				Folders:       []string{"workflows"},
+				SubjectPrefix: "cre-",
 			},
 		}, out.RedPanda.SchemaRegistryExternalURL)
 		require.NoError(t, err, "failed to register protos")

framework/observability/compose/conf/provisioning/dashboards/clnode-errors/errors.json

@@ -18,7 +18,6 @@
   "editable": true,
   "fiscalYearStartMonth": 0,
   "graphTooltip": 0,
-  "id": 4,
   "links": [],
   "liveNow": false,
   "panels": [
@@ -202,9 +201,10 @@
     },
     {
       "datasource": {
-        "type": "loki"
+        "type": "loki",
+        "uid": "P8E80F9AEF21F6940"
       },
-      "description": "Node Errors",
+      "description": "Node Criticals",
       "gridPos": {
         "h": 7,
         "w": 24,
@@ -223,6 +223,47 @@
         "wrapLogMessage": false
       },
       "pluginVersion": "10.2.6",
+      "targets": [
+        {
+          "datasource": {
+            "type": "loki"
+          },
+          "editorMode": "code",
+          "expr": "{job=\"ctf\", container=~\".*node.*\"}\n  | json \n  | level=\"crit\" \n  | logfmt \n  | line_format \"{{.container }} {{.logger }} {{.msg }} {{.err}}\"",
+          "legendFormat": "",
+          "queryType": "range",
+          "refId": "A"
+        }
+      ],
+      "title": "Node Criticals",
+      "transformations": [],
+      "transparent": true,
+      "type": "logs"
+    },
+    {
+      "datasource": {
+        "type": "loki",
+        "uid": "P8E80F9AEF21F6940"
+      },
+      "description": "Node Errors",
+      "gridPos": {
+        "h": 7,
+        "w": 24,
+        "x": 0,
+        "y": 17
+      },
+      "id": 5,
+      "options": {
+        "dedupStrategy": "none",
+        "enableLogDetails": true,
+        "prettifyLogMessage": false,
+        "showCommonLabels": false,
+        "showLabels": false,
+        "showTime": false,
+        "sortOrder": "Descending",
+        "wrapLogMessage": false
+      },
+      "pluginVersion": "10.2.6",
       "targets": [
         {
           "datasource": {
@@ -249,7 +290,7 @@
         "h": 6,
         "w": 24,
         "x": 0,
-        "y": 17
+        "y": 24
       },
       "id": 3,
       "options": {
@@ -280,7 +321,7 @@
       "type": "logs"
     }
   ],
-  "refresh": "",
+  "refresh": "5s",
   "schemaVersion": 38,
   "style": "dark",
   "tags": [],