EventProcessor Plugin Architecture

The EventProcessor's micro-plugin architecture supports a high degree of extensibility. With custom plugins, you can expand the system's capabilities in log ingestion, parsing, analysis, and notifications.

Architecture Highlights

CommunicationgRPC using Unix Domain Sockets.
Location: Sockets are automatically managed in {{WorkDir}}/sockets/.
Language: While gRPC supports many languages, we highly recommend Go and our official go-sdk.

Plugin Types

Type	SDK Helper	Interaction Pattern
Parsing	`InitParsingPlugin`	Converts logs inside a pipeline (unary RPC).
Analysis	`InitAnalysisPlugin`	Assesses events to generate alerts (server streaming).
Input	`SendLogsFromChannel`	Receives data from external sources and sends it into the engine.
Correlation	`InitCorrelationPlugin`	Analyzes alerts to identify high-level patterns.
Notification	`InitNotificationPlugin`	Sends messages to outside endpoints (Email, Slack, and the like).

🛠️ Quick Start (Go)

1. Initialize your project

mkdir synoti-plugin-custom && cd synoti-plugin-custom
go mod init github.com/user/synoti-plugin-custom
go get github.com/threatwinds/go-sdk

2. Build a Parsing Plugin

Parsing plugins handle specialized logic that standard steps (Grok, JSON, etc.) cannot accomplish.

package main

import (
	"context"
	"github.com/threatwinds/go-sdk/plugins"
	"github.com/tidwall/sjson"
)

func main() {
	// "my-enricher" results in socket: {{WorkDir}}/sockets/my-enricher_parsing.sock
	plugins.InitParsingPlugin("my-enricher", parseLog)
}

func parseLog(ctx context.Context, tr *plugins.Transform) (*plugins.Draft, error) {
	// Add custom field to the log
	newLog, _ := sjson.Set(tr.Draft.Log, "log.custom_status", "processed_by_plugin")
	tr.Draft.Log = newLog
	
	return tr.Draft, nil
}

3. Create an Input Plugin

Input plugins serve as producers. Instead of waiting for the engine to invoke them, they push data to the engine.

package main

import (
	"github.com/google/uuid"
	"github.com/threatwinds/go-sdk/plugins"
	"time"
)

func main() {
	pluginName := "my-data-source"
	
	// Start the background sender
	go plugins.SendLogsFromChannel(pluginName)

	for {
		// Simulate fetching data
		plugins.EnqueueLog(&plugins.Log{
			Id:         uuid.NewString(),
			DataType:   "custom_data",
			DataSource: "external_api",
			Timestamp:  time.Now().Format(time.RFC3339),
			Raw:        `{"message": "hello world"}`,
		}, pluginName)
		
		time.Sleep(10 * time.Second)
	}
}

🚀 Deployment & Configuration

Build: go build -o plugin-binary
Naming: The plugin name passed to InitParsingPlugin("name") has to correspond to your YAML configuration.

YAML Integration:

pipeline:
  - dataTypes: [custom_data]
    steps:
      - dynamic:
          plugin: my-enricher
          params: { key: value }

Best Practices

Use the Catcher: Always use github.com/threatwinds/go-sdk/catcher for reporting errors so logs stay standardized.
Reserved Fields: Ensure your parsing logic maps data to the common schema (origin.*, target.*, action). Do not create top-level fields beyond log.* unless they belong to the core schema.
Graceful Shutdown: The SDK's Init... functions automatically deal with SIGINT and SIGTERM.
Config Management: Use plugins.PluginCfg("plugin_name", true) to obtain configuration variables from the Unified Security Stack environment.