feat: OCSF Standardization Processor

Makefile

@@ -235,6 +235,7 @@ gosec:
 	  -exclude-dir=receiver/sapnetweaverreceiver \
 	  -exclude-dir=extension/bindplaneextension \
 	  -exclude-dir=processor/snapshotprocessor \
+	  -exclude-dir=processor/ocsfstandardizationprocessor \
 	  -exclude-dir=internal/tools \
 	  -exclude-dir=exporter/chronicleexporter/internal/metadata \
 	  -exclude-dir=exporter/chronicleexporter/protos/api \
@@ -247,6 +248,7 @@ gosec:
 	cd extension/bindplaneextension; gosec ./...
 	cd processor/snapshotprocessor; gosec ./...
 	cd receiver/sapnetweaverreceiver; gosec ./...
+	cd processor/ocsfstandardizationprocessor; gosec ./...
 
 # This target performs all checks that CI will do (excluding the build itself)
 .PHONY: ci-checks

processor/ocsfstandardizationprocessor/README.md

@@ -18,6 +18,7 @@ The following options may be configured:
 | -- | -- | -- | -- | -- |
 | `ocsf_version` | string | | Yes | The OCSF schema version. Supported: `1.0.0` through `1.7.0`. |
 | `event_mappings` | []EventMapping | `[]` | No | List of event mappings that define how logs are transformed. |
+| `runtime_validation` | bool | `true` | No | Enables runtime OCSF validation of mapped log bodies. When enabled, logs that do not conform to the OCSF schema (missing required fields, invalid enum values, regex/range constraint violations) are dropped. |
 
 ### EventMapping
 

processor/ocsfstandardizationprocessor/config.go

@@ -21,6 +21,14 @@ import (
 	"slices"
 
 	"github.com/observiq/bindplane-otel-collector/expr"
+	v100 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_0_0"
+	v110 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_1_0"
+	v120 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_2_0"
+	v130 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_3_0"
+	v140 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_4_0"
+	v150 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_5_0"
+	v160 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_6_0"
+	v170 "github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizationprocessor/ocsf/v1_7_0"
 )
 
 var (
@@ -73,8 +81,9 @@ type EventMapping struct {
 
 // Config is the configuration for the processor
 type Config struct {
-	OCSFVersion   OCSFVersion    `mapstructure:"ocsf_version"`
-	EventMappings []EventMapping `mapstructure:"event_mappings"`
+	OCSFVersion       OCSFVersion    `mapstructure:"ocsf_version"`
+	EventMappings     []EventMapping `mapstructure:"event_mappings"`
+	RuntimeValidation *bool          `mapstructure:"runtime_validation"`
 }
 
 // Validate validates the processor configuration
@@ -98,6 +107,11 @@ func (cfg Config) Validate() error {
 			}
 		}
 
+		defaultFieldCount := 2
+		fieldPaths := make([]string, len(em.FieldMappings)+defaultFieldCount)
+		// We always automatically add the class_uid field and the metadata.version field
+		fieldPaths[0] = "class_uid"
+		fieldPaths[1] = "metadata.version"
 		for j, fm := range em.FieldMappings {
 			if fm.To == "" {
 				return fmt.Errorf("event_mappings[%d].field_mappings[%d]: to is required", i, j)
@@ -111,6 +125,34 @@ func (cfg Config) Validate() error {
 					return fmt.Errorf("event_mappings[%d].field_mappings[%d]: invalid from expression: %w", i, j, err)
 				}
 			}
+
+			fieldPaths[j+defaultFieldCount] = fm.To
+		}
+
+		var coverageFunc func(classID int, fieldPaths []string) error
+		switch cfg.OCSFVersion {
+		case OCSFVersion1_0_0:
+			coverageFunc = v100.ValidateFieldCoverage
+		case OCSFVersion1_1_0:
+			coverageFunc = v110.ValidateFieldCoverage
+		case OCSFVersion1_2_0:
+			coverageFunc = v120.ValidateFieldCoverage
+		case OCSFVersion1_3_0:
+			coverageFunc = v130.ValidateFieldCoverage
+		case OCSFVersion1_4_0:
+			coverageFunc = v140.ValidateFieldCoverage
+		case OCSFVersion1_5_0:
+			coverageFunc = v150.ValidateFieldCoverage
+		case OCSFVersion1_6_0:
+			coverageFunc = v160.ValidateFieldCoverage
+		case OCSFVersion1_7_0:
+			coverageFunc = v170.ValidateFieldCoverage
+		default:
+			return fmt.Errorf("event_mappings[%d]: OCSF version %s is not supported", i, cfg.OCSFVersion)
+		}
+		coverageErr := coverageFunc(em.ClassID, fieldPaths)
+		if coverageErr != nil {
+			return fmt.Errorf("event_mappings[%d]: OCSF Class %d has validation errors\n%w", i, em.ClassID, coverageErr)
 		}
 	}
 

processor/ocsfstandardizationprocessor/config_test.go

@@ -20,6 +20,18 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// accountChangeFieldMappings provides the minimum required field mappings for
+// all versions of the AccountChange class (3001). class_uid and metadata.version are auto-added.
+var accountChangeFieldMappings = []FieldMapping{
+	{From: "body.activity", To: "activity_id"},
+	{From: "body.category", To: "category_uid"},
+	{From: "body.severity", To: "severity_id"},
+	{From: "body.time", To: "time"},
+	{From: "body.type", To: "type_uid"},
+	{From: "body.user", To: "user"},
+	{From: "body.product", To: "metadata.product"},
+}
+
 func TestConfigValidate(t *testing.T) {
 	tests := []struct {
 		name    string
@@ -29,14 +41,12 @@ func TestConfigValidate(t *testing.T) {
 		{
 			name: "valid config with all fields",
 			cfg: Config{
-				OCSFVersion: OCSFVersion1_3_0,
+				OCSFVersion: OCSFVersion1_0_0,
 				EventMappings: []EventMapping{
 					{
-						Filter:  "true",
-						ClassID: 1001,
-						FieldMappings: []FieldMapping{
-							{From: "body.src", To: "dst_endpoint.ip"},
-						},
+						Filter:        "true",
+						ClassID:       3001,
+						FieldMappings: accountChangeFieldMappings,
 					},
 				},
 			},
@@ -50,12 +60,18 @@ func TestConfigValidate(t *testing.T) {
 		{
 			name: "valid config with default value and from",
 			cfg: Config{
-				OCSFVersion: OCSFVersion1_7_0,
+				OCSFVersion: OCSFVersion1_0_0,
 				EventMappings: []EventMapping{
 					{
-						ClassID: 4001,
+						ClassID: 3001,
 						FieldMappings: []FieldMapping{
-							{From: "body.severity", To: "severity_id", Default: 1},
+							{From: "body.activity", To: "activity_id", Default: 1},
+							{From: "body.category", To: "category_uid"},
+							{From: "body.severity", To: "severity_id"},
+							{From: "body.time", To: "time"},
+							{From: "body.type", To: "type_uid"},
+							{From: "body.user", To: "user"},
+							{From: "body.product", To: "metadata.product"},
 						},
 					},
 				},
@@ -64,13 +80,11 @@ func TestConfigValidate(t *testing.T) {
 		{
 			name: "valid config with empty filter",
 			cfg: Config{
-				OCSFVersion: OCSFVersion1_3_0,
+				OCSFVersion: OCSFVersion1_0_0,
 				EventMappings: []EventMapping{
 					{
-						ClassID: 2001,
-						FieldMappings: []FieldMapping{
-							{From: "body.msg", To: "message"},
-						},
+						ClassID:       3001,
+						FieldMappings: accountChangeFieldMappings,
 					},
 				},
 			},
@@ -160,13 +174,11 @@ func TestConfigValidate(t *testing.T) {
 		{
 			name: "error in second event mapping",
 			cfg: Config{
-				OCSFVersion: OCSFVersion1_3_0,
+				OCSFVersion: OCSFVersion1_0_0,
 				EventMappings: []EventMapping{
 					{
-						ClassID: 1001,
-						FieldMappings: []FieldMapping{
-							{From: "body.src", To: "message"},
-						},
+						ClassID:       3001,
+						FieldMappings: accountChangeFieldMappings,
 					},
 					{
 						ClassID: 0,
@@ -204,3 +216,78 @@ func TestConfigValidate(t *testing.T) {
 		})
 	}
 }
+
+func TestConfigValidateFieldCoverage(t *testing.T) {
+	tests := []struct {
+		name          string
+		eventMappings []EventMapping
+		wantErr       string
+	}{
+		{
+			name: "unknown class UID",
+			eventMappings: []EventMapping{
+				{
+					ClassID: 9999,
+					FieldMappings: []FieldMapping{
+						{From: "body.msg", To: "message"},
+					},
+				},
+			},
+			wantErr: "OCSF Class 9999 has validation errors",
+		},
+		{
+			name: "missing required fields",
+			eventMappings: []EventMapping{
+				{
+					ClassID: 3001,
+					FieldMappings: []FieldMapping{
+						{From: "body.msg", To: "message"},
+					},
+				},
+			},
+			wantErr: "missing required field",
+		},
+		{
+			name: "all required fields covered",
+			eventMappings: []EventMapping{
+				{
+					ClassID:       3001,
+					FieldMappings: accountChangeFieldMappings,
+				},
+			},
+		},
+		{
+			name: "error in second event mapping",
+			eventMappings: []EventMapping{
+				{
+					ClassID:       3001,
+					FieldMappings: accountChangeFieldMappings,
+				},
+				{
+					ClassID: 9999,
+					FieldMappings: []FieldMapping{
+						{From: "body.msg", To: "message"},
+					},
+				},
+			},
+			wantErr: "OCSF Class 9999 has validation errors",
+		},
+	}
+
+	for _, version := range OCSFVersions {
+		for _, tt := range tests {
+			t.Run(string(version)+"/"+tt.name, func(t *testing.T) {
+				cfg := Config{
+					OCSFVersion:   version,
+					EventMappings: tt.eventMappings,
+				}
+				err := cfg.Validate()
+				if tt.wantErr != "" {
+					require.ErrorContains(t, err, tt.wantErr)
+				} else {
+					require.NoError(t, err)
+				}
+			})
+		}
+	}
+}

processor/ocsfstandardizationprocessor/factory.go

@@ -25,7 +25,8 @@ import (
 )
 
 var (
-	errInvalidConfigType = errors.New("config is not of type removeemptyvaluesprocessor.Config")
+	// errInvalidConfigType is returned when the configuration is not of the expected type.
+	errInvalidConfigType = errors.New("config is not of type ocsfstandardizationprocessor.Config")
 	// componentType is the value of the "type" key in configuration.
 	componentType = component.MustNewType("ocsf_standardization")
 )

processor/ocsfstandardizationprocessor/go.mod

@@ -3,7 +3,6 @@ module github.com/observiq/bindplane-otel-collector/processor/ocsfstandardizatio
 go 1.25.7
 
 require (
-	github.com/go-viper/mapstructure/v2 v2.5.0
 	github.com/observiq/bindplane-otel-collector/expr v0.0.0-00010101000000-000000000000
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/collector/component v1.52.0

processor/ocsfstandardizationprocessor/go.sum

@@ -23,8 +23,6 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
 github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
-github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro=
-github.com/go-viper/mapstructure/v2 v2.5.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
 github.com/gobwas/glob v0.2.3 h1:A4xDbljILXROh+kObIiy5kIaPYD8e96x1tgBhUI5J+Y=
 github.com/gobwas/glob v0.2.3/go.mod h1:d3Ez4x06l9bZtSvzIay5+Yzi0fmZzPgnTbPcKjJAkT8=
 github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4=