diff --git a/articles/log-destinations.md b/articles/log-destinations.md index f4b92543c2..abbb46ac8f 100644 --- a/articles/log-destinations.md +++ b/articles/log-destinations.md @@ -36,7 +36,30 @@ Snowflake provides instructions on setting up the destination tables and IAM rol ## Splunk -How to send logs to Splunk: +Logs are sent directly to [Splunk](https://www.splunk.com/) via the [HTTP Event Collector (HEC)](https://docs.splunk.com/Documentation/Splunk/latest/Data/UsetheHTTPEventCollector) endpoint. + +- Plugin name: `splunk` +- Flag namespace: [splunk](https://fleetdm.com/docs/deploying/configuration#splunk) + +Events are batched up to 1MB before sending. Events over 1MB are dropped, with a notification sent to the Fleet server logs. Fleet retries on transient errors (HTTP 503) with exponential backoff. + +To use this destination, enable HEC on your Splunk instance and create an HEC token. Then configure Fleet with the HEC URL and token: + +```yaml +osquery: + status_log_plugin: splunk + result_log_plugin: splunk +splunk: + url: https://splunk.example.com:8088 + token: your-hec-token + index: main + source: fleet + source_type: fleet:json +``` + +### Splunk via Firehose (alternative) + +You can also send logs to Splunk indirectly through Amazon Kinesis Data Firehose: 1. Follow [Splunk's instructions](https://docs.splunk.com/Documentation/AddOns/latest/Firehose/ConfigureFirehose) to prepare Splunk for Firehose data. diff --git a/changes/25574-splunk-log-destination b/changes/25574-splunk-log-destination new file mode 100644 index 0000000000..e941a94dd8 --- /dev/null +++ b/changes/25574-splunk-log-destination @@ -0,0 +1 @@ +- Added native Splunk HEC log destination for osquery status, result, and audit logs. diff --git a/cmd/fleet/logging.go b/cmd/fleet/logging.go index b8bed08b59..0380907c3d 100644 --- a/cmd/fleet/logging.go +++ b/cmd/fleet/logging.go @@ -66,6 +66,14 @@ func buildLoggingConfig(cfg config.FleetConfig) logging.Config { JetStream: cfg.Nats.JetStream, Timeout: cfg.Nats.Timeout, }, + Splunk: logging.SplunkConfig{ + URL: cfg.Splunk.URL, + Token: cfg.Splunk.Token, + Index: cfg.Splunk.Index, + Source: cfg.Splunk.Source, + SourceType: cfg.Splunk.SourceType, + InsecureSkipVerify: cfg.Splunk.InsecureSkipVerify, + }, } } diff --git a/docs/Configuration/fleet-server-configuration.md b/docs/Configuration/fleet-server-configuration.md index ef218b0133..686cda533f 100644 --- a/docs/Configuration/fleet-server-configuration.md +++ b/docs/Configuration/fleet-server-configuration.md @@ -1169,7 +1169,7 @@ Valid time units are `s`, `m`, `h`. This is the log output plugin that should be used for osquery status logs received from clients. Check out the [reference documentation for log destinations](https://fleetdm.com/docs/using-fleet/log-destinations). -Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, `nats`, and `stdout`. +Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, `nats`, `splunk`, and `stdout`. - Default value: `filesystem` - Environment variable: `FLEET_OSQUERY_STATUS_LOG_PLUGIN` @@ -1183,7 +1183,7 @@ Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest` This is the log output plugin that should be used for osquery result logs received from clients. Check out the [reference documentation for log destinations](https://fleetdm.com/docs/using-fleet/log-destinations). -Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, `nats`, and `stdout`. +Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, `nats`, `splunk`, and `stdout`. - Default value: `filesystem` - Environment variable: `FLEET_OSQUERY_RESULT_LOG_PLUGIN` @@ -1427,7 +1427,7 @@ This flag only has effect if `activity_enable_audit_log` is set to `true`. Each plugin has additional configuration options. Please see the configuration section linked below for your logging plugin. -Options are [`filesystem`](#filesystem), [`firehose`](#firehose), [`kinesis`](#kinesis), [`lambda`](#lambda), [`pubsub`](#pubsub), [`kafkarest`](#kafka-rest-proxy-logging), [`nats`](#nats), and `stdout` (no additional configuration needed). +Options are [`filesystem`](#filesystem), [`firehose`](#firehose), [`kinesis`](#kinesis), [`lambda`](#lambda), [`pubsub`](#pubsub), [`kafkarest`](#kafka-rest-proxy-logging), [`nats`](#nats), [`splunk`](#splunk), and `stdout` (no additional configuration needed). - Default value: `filesystem` - Environment variable: `FLEET_ACTIVITY_AUDIT_LOG_PLUGIN` @@ -2527,6 +2527,106 @@ Timeout for NATS publish operations. Valid time units are `s`, `m`, `h`. timeout: 1m ``` +## Splunk + +Fleet can send osquery logs directly to Splunk via the [HTTP Event Collector (HEC)](https://docs.splunk.com/Documentation/Splunk/latest/Data/UsetheHTTPEventCollector) endpoint. + +### splunk_url + +This flag only has effect if one of the following is true: +- `osquery_result_log_plugin` or `osquery_status_log_plugin` are set to `splunk`. +- `activity_audit_log_plugin` is set to `splunk` and `activity_enable_audit_log` is set to `true`. + +The base URL of the Splunk HEC endpoint (e.g. `https://splunk.example.com:8088`). + +- Default value: none +- Environment variable: `FLEET_SPLUNK_URL` +- Config file format: + ```yaml + splunk: + url: https://splunk.example.com:8088 + ``` + +### splunk_token + +This flag only has effect if one of the following is true: +- `osquery_result_log_plugin` or `osquery_status_log_plugin` are set to `splunk`. +- `activity_audit_log_plugin` is set to `splunk` and `activity_enable_audit_log` is set to `true`. + +The HEC authentication token. + +- Default value: none +- Environment variable: `FLEET_SPLUNK_TOKEN` +- Config file format: + ```yaml + splunk: + token: your-hec-token + ``` + +### splunk_index + +This flag only has effect if one of the following is true: +- `osquery_result_log_plugin` or `osquery_status_log_plugin` are set to `splunk`. +- `activity_audit_log_plugin` is set to `splunk` and `activity_enable_audit_log` is set to `true`. + +The Splunk index to send events to. If empty, the HEC token's default index is used. + +- Default value: none +- Environment variable: `FLEET_SPLUNK_INDEX` +- Config file format: + ```yaml + splunk: + index: main + ``` + +### splunk_source + +This flag only has effect if one of the following is true: +- `osquery_result_log_plugin` or `osquery_status_log_plugin` are set to `splunk`. +- `activity_audit_log_plugin` is set to `splunk` and `activity_enable_audit_log` is set to `true`. + +The source value for events sent to Splunk. If empty, the HEC token's default source is used. + +- Default value: none +- Environment variable: `FLEET_SPLUNK_SOURCE` +- Config file format: + ```yaml + splunk: + source: fleet + ``` + +### splunk_source_type + +This flag only has effect if one of the following is true: +- `osquery_result_log_plugin` or `osquery_status_log_plugin` are set to `splunk`. +- `activity_audit_log_plugin` is set to `splunk` and `activity_enable_audit_log` is set to `true`. + +The sourcetype value for events sent to Splunk. If empty, the HEC token's default sourcetype is used. + +- Default value: none +- Environment variable: `FLEET_SPLUNK_SOURCE_TYPE` +- Config file format: + ```yaml + splunk: + source_type: fleet:json + ``` + +### splunk_insecure_skip_verify + +This flag only has effect if one of the following is true: +- `osquery_result_log_plugin` or `osquery_status_log_plugin` are set to `splunk`. +- `activity_audit_log_plugin` is set to `splunk` and `activity_enable_audit_log` is set to `true`. + +Skip TLS certificate verification when connecting to the Splunk HEC endpoint. Useful for development environments with self-signed certificates. + +- Default value: `false` +- Environment variable: `FLEET_SPLUNK_INSECURE_SKIP_VERIFY` +- Config file format: + ```yaml + splunk: + insecure_skip_verify: true + ``` + ## Email backend By default, the SMTP backend is enabled and no additional configuration is required on the server settings. You can configure diff --git a/docs/Get started/FAQ.md b/docs/Get started/FAQ.md index ea9a7ebbb5..4dda82a941 100644 --- a/docs/Get started/FAQ.md +++ b/docs/Get started/FAQ.md @@ -274,7 +274,7 @@ For results to go to Fleet, the osquery `--logger_plugin` flag must be set to `t Folks typically use Fleet to ship logs to data lakes and SIEMs like Splunk, the ELK stack, and Graylog. Fleet supports multiple logging destinations for scheduled query results and status logs. The `--osquery_result_log_plugin` and `--osquery_status_log_plugin` can be set to: -`filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, `nats`, and `stdout`. +`filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, `nats`, `splunk`, and `stdout`. See: - https://fleetdm.com/docs/deploying/configuration#osquery-result-log-plugin. - https://fleetdm.com/docs/deploying/configuration#osquery-status-log-plugin. diff --git a/frontend/components/LogDestinationIndicator/LogDestinationIndicator.tsx b/frontend/components/LogDestinationIndicator/LogDestinationIndicator.tsx index a05fdaefef..3d239f33aa 100644 --- a/frontend/components/LogDestinationIndicator/LogDestinationIndicator.tsx +++ b/frontend/components/LogDestinationIndicator/LogDestinationIndicator.tsx @@ -46,6 +46,8 @@ const LogDestinationIndicator = ({ return "Apache Kafka"; case "nats": return "NATS"; + case "splunk": + return "Splunk"; case "stdout": return "Standard output (stdout)"; case "webhook": @@ -107,6 +109,12 @@ const LogDestinationIndicator = ({ Each time a report runs, the data
is sent to NATS. ); + case "splunk": + return ( + <> + Each time a report runs, the data
is sent to Splunk. + + ); case "stdout": return ( <> diff --git a/frontend/interfaces/config.ts b/frontend/interfaces/config.ts index ead128a9bf..8a68ef3fb9 100644 --- a/frontend/interfaces/config.ts +++ b/frontend/interfaces/config.ts @@ -267,6 +267,7 @@ export type LogDestination = | "pubsub" | "kafka" | "nats" + | "splunk" | "stdout" | "webhook" | ""; diff --git a/server/config/config.go b/server/config/config.go index 7d7a6c6cf4..b8ec88edac 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -687,6 +687,16 @@ type KafkaRESTConfig struct { Timeout int `json:"timeout" yaml:"timeout"` } +// SplunkConfig defines configs for the Splunk HEC logging plugin. +type SplunkConfig struct { + URL string `json:"url" yaml:"url"` + Token string `json:"token" yaml:"token"` + Index string `json:"index" yaml:"index"` + Source string `json:"source" yaml:"source"` + SourceType string `json:"source_type" yaml:"source_type"` + InsecureSkipVerify bool `json:"insecure_skip_verify" yaml:"insecure_skip_verify"` +} + // NatsConfig defines configs for the NATS logging plugin. type NatsConfig struct { StatusSubject string `json:"status_subject" yaml:"status_subject"` @@ -799,6 +809,7 @@ type FleetConfig struct { Webhook WebhookConfig KafkaREST KafkaRESTConfig Nats NatsConfig + Splunk SplunkConfig License LicenseConfig Vulnerabilities VulnerabilitiesConfig Upgrades UpgradesConfig @@ -1689,6 +1700,14 @@ func (man Manager) addConfigs() { man.addConfigBool("nats.jetstream", false, "NATS JetStream publish") man.addConfigDuration("nats.timeout", 30*time.Second, "NATS timeout") + // Splunk + man.addConfigString("splunk.url", "", "Splunk HEC URL (e.g. https://splunk.example.com:8088)") + man.addConfigString("splunk.token", "", "Splunk HEC authentication token") + man.addConfigString("splunk.index", "", "Splunk index to send events to") + man.addConfigString("splunk.source", "", "Splunk source value for events") + man.addConfigString("splunk.source_type", "", "Splunk sourcetype value for events") + man.addConfigBool("splunk.insecure_skip_verify", false, "Skip TLS certificate verification for Splunk HEC (for self-signed certs)") + // License man.addConfigString("license.key", "", "Fleet license key (to enable Fleet Premium features)") man.addConfigBool("license.enforce_host_limit", false, "Enforce license limit of enrolled hosts") @@ -2062,6 +2081,14 @@ func (man Manager) LoadConfig() FleetConfig { JetStream: man.getConfigBool("nats.jetstream"), Timeout: man.getConfigDuration("nats.timeout"), }, + Splunk: SplunkConfig{ + URL: man.getConfigString("splunk.url"), + Token: man.getConfigString("splunk.token"), + Index: man.getConfigString("splunk.index"), + Source: man.getConfigString("splunk.source"), + SourceType: man.getConfigString("splunk.source_type"), + InsecureSkipVerify: man.getConfigBool("splunk.insecure_skip_verify"), + }, License: LicenseConfig{ Key: man.getConfigString("license.key"), EnforceHostLimit: man.getConfigBool("license.enforce_host_limit"), diff --git a/server/fleet/app.go b/server/fleet/app.go index 21b9b1ebb7..00bb65fd92 100644 --- a/server/fleet/app.go +++ b/server/fleet/app.go @@ -2056,6 +2056,14 @@ type NatsConfig struct { AuditSubject string `json:"audit_subject"` } +// SplunkConfig shadows config.SplunkConfig only exposing a subset of fields +type SplunkConfig struct { + URL string `json:"url"` + Index string `json:"index"` + Source string `json:"source"` + SourceType string `json:"source_type"` +} + // DeviceGlobalConfig is a subset of AppConfig with information used by the // device endpoints type DeviceGlobalConfig struct { diff --git a/server/logging/logging.go b/server/logging/logging.go index 0ef803f4c7..67637a8cae 100644 --- a/server/logging/logging.go +++ b/server/logging/logging.go @@ -88,6 +88,15 @@ type NatsConfig struct { Timeout time.Duration } +type SplunkConfig struct { + URL string + Token string + Index string + Source string + SourceType string + InsecureSkipVerify bool +} + type Config struct { Plugin string @@ -99,6 +108,7 @@ type Config struct { PubSub PubSubConfig KafkaREST KafkaRESTConfig Nats NatsConfig + Splunk SplunkConfig } func NewJSONLogger(ctx context.Context, name string, config Config, logger *slog.Logger) (fleet.JSONLogger, error) { @@ -220,6 +230,26 @@ func NewJSONLogger(ctx context.Context, name string, config Config, logger *slog return nil, fmt.Errorf("create nats %s logger: %w", name, err) } return fleet.JSONLogger(writer), nil + case "splunk": + if config.Splunk.URL == "" { + return nil, fmt.Errorf("splunk %s logger: URL must not be empty", name) + } + if config.Splunk.Token == "" { + return nil, fmt.Errorf("splunk %s logger: HEC token must not be empty", name) + } + writer, err := NewSplunkLogWriter( + config.Splunk.URL, + config.Splunk.Token, + config.Splunk.Index, + config.Splunk.Source, + config.Splunk.SourceType, + config.Splunk.InsecureSkipVerify, + logger, + ) + if err != nil { + return nil, fmt.Errorf("create splunk %s logger: %w", name, err) + } + return fleet.JSONLogger(writer), nil default: return nil, fmt.Errorf( "unknown %s log plugin: %s", name, config.Plugin, diff --git a/server/logging/splunk.go b/server/logging/splunk.go new file mode 100644 index 0000000000..a8039bcf3a --- /dev/null +++ b/server/logging/splunk.go @@ -0,0 +1,193 @@ +package logging + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "time" + + "github.com/fleetdm/fleet/v4/pkg/fleethttp" + "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" +) + +const ( + // splunkHECPath is the Splunk HTTP Event Collector endpoint. + splunkHECPath = "/services/collector/event" + // splunkHealthPath is the HEC health check endpoint. + splunkHealthPath = "/services/collector/health" + // splunkMaxBatchSize is the default max content length for HEC (1 MB). + splunkMaxBatchSize = 1_000_000 + // splunkMaxSizeOfRecord is the max size of a single HEC event (1 MB). + splunkMaxSizeOfRecord = 1_000_000 + // splunkMaxRetries is the maximum number of retries on transient errors. + splunkMaxRetries = 8 +) + +// splunkEvent wraps a log entry in the Splunk HEC event format. +type splunkEvent struct { + Event json.RawMessage `json:"event"` + // Time is the event timestamp in epoch seconds. + Time float64 `json:"time,omitempty"` + // Index is the Splunk index to send events to. + Index string `json:"index,omitempty"` + // Source overrides the default source. + Source string `json:"source,omitempty"` + // SourceType overrides the default sourcetype. + SourceType string `json:"sourcetype,omitempty"` +} + +type splunkLogWriter struct { + url string + token string + index string + source string + sourceType string + client *http.Client + logger *slog.Logger +} + +func NewSplunkLogWriter(url, token, index, source, sourceType string, insecureSkipVerify bool, logger *slog.Logger) (*splunkLogWriter, error) { + clientOpts := []fleethttp.ClientOpt{fleethttp.WithTimeout(30 * time.Second)} + if insecureSkipVerify { + clientOpts = append(clientOpts, fleethttp.WithTLSClientConfig(&tls.Config{ + InsecureSkipVerify: true, //nolint:gosec // user-configured option for self-signed certs + })) + } + + w := &splunkLogWriter{ + url: url, + token: token, + index: index, + source: source, + sourceType: sourceType, + client: fleethttp.NewClient(clientOpts...), + logger: logger, + } + + if err := w.checkHealth(); err != nil { + return nil, fmt.Errorf("splunk health check: %w", err) + } + + return w, nil +} + +func (w *splunkLogWriter) Write(ctx context.Context, logs []json.RawMessage) error { + if len(logs) == 0 { + return nil + } + + now := float64(time.Now().UnixNano()) / float64(time.Second) + + var buf bytes.Buffer + for _, l := range logs { + evt := splunkEvent{ + Event: l, + Time: now, + Index: w.index, + Source: w.source, + SourceType: w.sourceType, + } + b, err := json.Marshal(evt) + if err != nil { + w.logger.ErrorContext(ctx, "failed to marshal splunk event", "err", err) + continue + } + + if len(b) > splunkMaxSizeOfRecord { + w.logger.InfoContext(ctx, "dropping splunk event over 1MB limit", + "size", len(b), + ) + continue + } + + // If adding this event would exceed the batch size, flush first. + if buf.Len() > 0 && buf.Len()+len(b) > splunkMaxBatchSize { + if err := w.send(ctx, buf.Bytes()); err != nil { + return err + } + buf.Reset() + } + + buf.Write(b) + } + + if buf.Len() > 0 { + return w.send(ctx, buf.Bytes()) + } + + return nil +} + +func (w *splunkLogWriter) send(ctx context.Context, payload []byte) error { + return w.sendWithRetry(ctx, payload, 0) +} + +// splunkRetryDelay calculates the backoff duration for a given retry attempt. +// Exported as a var so tests can override it to avoid waiting. +var splunkRetryDelay = func(try int) time.Duration { + return 100 * time.Millisecond * time.Duration(1< 0 { + timer := time.NewTimer(splunkRetryDelay(try)) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctxerr.Wrap(ctx, ctx.Err(), "splunk retry canceled") + case <-timer.C: + } + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, w.url+splunkHECPath, bytes.NewReader(payload)) + if err != nil { + return ctxerr.Wrap(ctx, err, "splunk create request") + } + req.Header.Set("Authorization", "Splunk "+w.token) + req.Header.Set("Content-Type", "application/json") + + resp, err := w.client.Do(req) + if err != nil { + return ctxerr.Wrap(ctx, err, "splunk send") + } + defer resp.Body.Close() + + if (resp.StatusCode == http.StatusServiceUnavailable || resp.StatusCode == http.StatusTooManyRequests) && try < splunkMaxRetries { + io.Copy(io.Discard, resp.Body) //nolint:errcheck // best-effort drain for connection reuse + resp.Body.Close() + return w.sendWithRetry(ctx, payload, try+1) + } + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return ctxerr.Errorf(ctx, "splunk HEC returned status %d: %s", resp.StatusCode, string(body)) + } + + return nil +} + +func (w *splunkLogWriter) checkHealth() error { + req, err := http.NewRequest(http.MethodGet, w.url+splunkHealthPath, nil) + if err != nil { + return fmt.Errorf("create health request: %w", err) + } + req.Header.Set("Authorization", "Splunk "+w.token) + + resp, err := w.client.Do(req) + if err != nil { + return fmt.Errorf("health request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("HEC health check returned status %d: %s", resp.StatusCode, string(body)) + } + + return nil +} diff --git a/server/logging/splunk_integration_test.go b/server/logging/splunk_integration_test.go new file mode 100644 index 0000000000..0d78933dbb --- /dev/null +++ b/server/logging/splunk_integration_test.go @@ -0,0 +1,162 @@ +package logging + +import ( + "bytes" + "crypto/tls" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "os" + "testing" + "time" + + "github.com/fleetdm/fleet/v4/pkg/fleethttp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestSplunkIntegration tests the Splunk HEC writer against a real Splunk instance. +// +// Prerequisites: +// +// docker run -d --name splunk-test --platform linux/amd64 \ +// -p 8000:8000 -p 8088:8088 -p 8089:8089 \ +// -e SPLUNK_GENERAL_TERMS=--accept-sgt-current-at-splunk-com \ +// -e SPLUNK_START_ARGS=--accept-license \ +// -e SPLUNK_PASSWORD=changeme123 \ +// -e SPLUNK_HEC_TOKEN=test-hec-token-1234 \ +// splunk/splunk:latest +// +// Run with: SPLUNK_INTEGRATION_TEST=1 go test ./server/logging/ -run TestSplunkIntegration -v +func TestSplunkIntegration(t *testing.T) { + if os.Getenv("SPLUNK_INTEGRATION_TEST") == "" { + t.Skip("set SPLUNK_INTEGRATION_TEST=1 to run this test (requires a running Splunk instance)") + } + + splunkURL := "https://localhost:8088" + splunkToken := "test-hec-token-1234" + + ctx := t.Context() + + // 1. Create writer with insecureSkipVerify for the self-signed cert + writer, err := NewSplunkLogWriter(splunkURL, splunkToken, "main", "fleet-integration-test", "fleet:json", true, slog.Default()) + require.NoError(t, err, "NewSplunkLogWriter should connect to the running Splunk instance") + + // 2. Send test events + marker := fmt.Sprintf("integration-test-%d", time.Now().UnixNano()) + testLogs := []json.RawMessage{ + json.RawMessage(fmt.Sprintf(`{"marker":"%s","seq":1,"host":"test-host-1","status":"ok"}`, marker)), + json.RawMessage(fmt.Sprintf(`{"marker":"%s","seq":2,"host":"test-host-2","status":"warning"}`, marker)), + json.RawMessage(fmt.Sprintf(`{"marker":"%s","seq":3,"host":"test-host-3","status":"error"}`, marker)), + } + + err = writer.Write(ctx, testLogs) + require.NoError(t, err, "Write should send events to Splunk HEC without error") + + // 3. Query Splunk REST API to verify events landed. + // Give Splunk a moment to index the events. + time.Sleep(5 * time.Second) + + events := searchSplunk(t, marker) + require.Len(t, events, 3, "should find all 3 test events in Splunk") + + // Verify event content (Splunk returns newest first) + for _, evt := range events { + assert.Contains(t, evt, marker, "event should contain our unique marker") + } + +} + +// TestSplunkIntegrationBatch tests batch splitting against a real Splunk instance. +func TestSplunkIntegrationBatch(t *testing.T) { + if os.Getenv("SPLUNK_INTEGRATION_TEST") == "" { + t.Skip("set SPLUNK_INTEGRATION_TEST=1 to run this test (requires a running Splunk instance)") + } + + splunkURL := "https://localhost:8088" + splunkToken := "test-hec-token-1234" + + ctx := t.Context() + + writer, err := NewSplunkLogWriter(splunkURL, splunkToken, "main", "fleet-batch-test", "fleet:json", true, slog.Default()) + require.NoError(t, err) + + // Send 100 events to verify batching works + marker := fmt.Sprintf("batch-test-%d", time.Now().UnixNano()) + testLogs := make([]json.RawMessage, 100) + for i := range testLogs { + testLogs[i] = json.RawMessage(fmt.Sprintf(`{"marker":"%s","seq":%d,"data":"%s"}`, marker, i, "payload-data-for-batch-test")) + } + + err = writer.Write(ctx, testLogs) + require.NoError(t, err, "Write should handle batch of 100 events") + + time.Sleep(5 * time.Second) + + events := searchSplunk(t, marker) + require.Len(t, events, 100, "all 100 events should be indexed in Splunk") + +} + +// TestSplunkIntegrationBadToken tests that sending with a bad token is rejected by HEC. +// Note: the HEC /health endpoint returns 200 regardless of token validity (it reports +// overall HEC health), so token validation only happens on the event endpoint. +func TestSplunkIntegrationBadToken(t *testing.T) { + if os.Getenv("SPLUNK_INTEGRATION_TEST") == "" { + t.Skip("set SPLUNK_INTEGRATION_TEST=1 to run this test (requires a running Splunk instance)") + } + + splunkURL := "https://localhost:8088" + ctx := t.Context() + + // Health check passes (it doesn't validate tokens), but Write should fail. + writer, err := NewSplunkLogWriter(splunkURL, "bad-token-12345", "main", "fleet", "fleet:json", true, slog.Default()) + require.NoError(t, err, "health check passes regardless of token") + + err = writer.Write(ctx, []json.RawMessage{json.RawMessage(`{"test":"bad-token"}`)}) + require.Error(t, err, "Write should fail with an invalid token") + require.Contains(t, err.Error(), "403") +} + +// searchSplunk queries the Splunk REST API for events containing the given marker string. +func searchSplunk(t *testing.T, marker string) []string { + t.Helper() + + searchQuery := fmt.Sprintf(`search index=main "%s" | fields _raw`, marker) + client := fleethttp.NewClient(fleethttp.WithTLSClientConfig(&tls.Config{ + InsecureSkipVerify: true, //nolint:gosec // test-only, local Docker Splunk + })) + + body := fmt.Sprintf("search=%s&output_mode=json&earliest_time=-5m", searchQuery) + req, err := http.NewRequest(http.MethodPost, "https://localhost:8089/services/search/jobs/export", bytes.NewBufferString(body)) + require.NoError(t, err) + req.SetBasicAuth("admin", "changeme123") + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + resp, err := client.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Equal(t, http.StatusOK, resp.StatusCode, "Splunk search API returned: %s", string(respBody)) + + // Parse the NDJSON response (one JSON object per line). + var events []string + dec := json.NewDecoder(bytes.NewReader(respBody)) + for dec.More() { + var result map[string]any + if err := dec.Decode(&result); err != nil { + break + } + if raw, ok := result["result"].(map[string]any); ok { + if rawStr, ok := raw["_raw"].(string); ok { + events = append(events, rawStr) + } + } + } + + return events +} diff --git a/server/logging/splunk_test.go b/server/logging/splunk_test.go new file mode 100644 index 0000000000..101e45a282 --- /dev/null +++ b/server/logging/splunk_test.go @@ -0,0 +1,315 @@ +package logging + +import ( + "bytes" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSplunkWrite(t *testing.T) { + ctx := t.Context() + + var receivedBody []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + assert.Equal(t, splunkHECPath, r.URL.Path) + assert.Equal(t, "Splunk test-token", r.Header.Get("Authorization")) + assert.Equal(t, "application/json", r.Header.Get("Content-Type")) + + var err error + receivedBody, err = io.ReadAll(r.Body) + assert.NoError(t, err) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "main", "fleet", "fleet:json", false, slog.Default()) + require.NoError(t, err) + + err = writer.Write(ctx, logs) + require.NoError(t, err) + require.NotEmpty(t, receivedBody) + + // The body should be concatenated JSON objects (one per log entry). + decoder := json.NewDecoder(bytes.NewReader(receivedBody)) + var events []splunkEvent + for decoder.More() { + var evt splunkEvent + err := decoder.Decode(&evt) + require.NoError(t, err) + events = append(events, evt) + } + + require.Len(t, events, 3) + for i, evt := range events { + assert.JSONEq(t, string(logs[i]), string(evt.Event)) + assert.Equal(t, "main", evt.Index) + assert.Equal(t, "fleet", evt.Source) + assert.Equal(t, "fleet:json", evt.SourceType) + assert.NotZero(t, evt.Time) + } +} + +func TestSplunkWriteEmpty(t *testing.T) { + ctx := t.Context() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + t.Fatal("should not send request for empty logs") + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + err = writer.Write(ctx, []json.RawMessage{}) + require.NoError(t, err) +} + +func TestSplunkServerError(t *testing.T) { + ctx := t.Context() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + http.Error(w, `{"text":"Invalid token","code":4}`, http.StatusForbidden) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + err = writer.Write(ctx, logs) + require.Error(t, err) + require.Contains(t, err.Error(), "403") +} + +func TestSplunkHealthCheckFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "Service Unavailable", http.StatusServiceUnavailable) + })) + defer server.Close() + + _, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "health check") +} + +func TestSplunkRecordTooBig(t *testing.T) { + ctx := t.Context() + + var receivedBody []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + var err error + receivedBody, err = io.ReadAll(r.Body) + assert.NoError(t, err) + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + // Create one normal log and one oversized log (>1MB) + normalLog := json.RawMessage(`{"normal":"event"}`) + bigPayload := make([]byte, splunkMaxSizeOfRecord+1) + for i := range bigPayload { + bigPayload[i] = 'x' + } + oversizedLog := json.RawMessage(`{"big":"` + string(bigPayload) + `"}`) + + err = writer.Write(ctx, []json.RawMessage{normalLog, oversizedLog}) + require.NoError(t, err) + + // Only the normal event should have been sent; the oversized one should be dropped + decoder := json.NewDecoder(bytes.NewReader(receivedBody)) + var count int + for decoder.More() { + var evt splunkEvent + err := decoder.Decode(&evt) + require.NoError(t, err) + count++ + } + assert.Equal(t, 1, count, "only the normal-sized event should be sent") +} + +func TestSplunkSplitBatchBySize(t *testing.T) { + ctx := t.Context() + + var batchCount int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + batchCount++ + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + // Create logs that together exceed splunkMaxBatchSize (1MB). + // Each log wraps to ~10KB after HEC envelope, so ~120 logs should exceed 1MB. + var largeLogs []json.RawMessage + payload := make([]byte, 10000) + for i := range payload { + payload[i] = 'a' + } + for range 120 { + largeLogs = append(largeLogs, json.RawMessage(`{"data":"`+string(payload)+`"}`)) + } + + err = writer.Write(ctx, largeLogs) + require.NoError(t, err) + assert.Greater(t, batchCount, 1, "should split into multiple batches") +} + +func TestSplunkRetryOnServiceUnavailable(t *testing.T) { + ctx := t.Context() + origDelay := splunkRetryDelay + splunkRetryDelay = func(_ int) time.Duration { return time.Millisecond } + t.Cleanup(func() { splunkRetryDelay = origDelay }) + + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + callCount++ + if callCount <= 2 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + err = writer.Write(ctx, logs) + require.NoError(t, err) + assert.Equal(t, 3, callCount, "should retry twice then succeed on third attempt") +} + +func TestSplunkRetryExhausted(t *testing.T) { + ctx := t.Context() + origDelay := splunkRetryDelay + splunkRetryDelay = func(_ int) time.Duration { return time.Millisecond } + t.Cleanup(func() { splunkRetryDelay = origDelay }) + + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + callCount++ + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + err = writer.Write(ctx, logs) + require.Error(t, err) + require.Contains(t, err.Error(), "503") + // 1 initial attempt + 8 retries = 9 total + assert.Equal(t, splunkMaxRetries+1, callCount, "should exhaust all retries") +} + +func TestSplunkRetryBodyIntegrity(t *testing.T) { + ctx := t.Context() + origDelay := splunkRetryDelay + splunkRetryDelay = func(_ int) time.Duration { return time.Millisecond } + t.Cleanup(func() { splunkRetryDelay = origDelay }) + + var bodies [][]byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + b, _ := io.ReadAll(r.Body) + bodies = append(bodies, b) + if len(bodies) <= 2 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + err = writer.Write(ctx, logs) + require.NoError(t, err) + require.Len(t, bodies, 3) + // Every retry must send the exact same payload + assert.Equal(t, bodies[0], bodies[1], "retry 1 body must match original") + assert.Equal(t, bodies[0], bodies[2], "retry 2 body must match original") + assert.NotEmpty(t, bodies[0], "body must not be empty") +} + +func TestSplunkRetryNoNestedRetries(t *testing.T) { + ctx := t.Context() + origDelay := splunkRetryDelay + splunkRetryDelay = func(_ int) time.Duration { return time.Millisecond } + t.Cleanup(func() { splunkRetryDelay = origDelay }) + + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == splunkHealthPath { + w.WriteHeader(http.StatusOK) + return + } + callCount++ + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer server.Close() + + writer, err := NewSplunkLogWriter(server.URL, "test-token", "", "", "", false, slog.Default()) + require.NoError(t, err) + + _ = writer.Write(ctx, logs) + // Must be exactly splunkMaxRetries+1, not exponentially more. + // Nested retries would produce 2^9 = 512 calls. + assert.Equal(t, splunkMaxRetries+1, callCount, "retries must be linear, not nested") +} + +func TestSplunkMissingConfig(t *testing.T) { + ctx := t.Context() + // Validation now happens in the factory (logging.go), not in NewSplunkLogWriter. + _, err := NewJSONLogger(ctx, "status", Config{Plugin: "splunk", Splunk: SplunkConfig{Token: "t"}}, slog.Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "URL") + + _, err = NewJSONLogger(ctx, "status", Config{Plugin: "splunk", Splunk: SplunkConfig{URL: "http://localhost"}}, slog.Default()) + require.Error(t, err) + require.Contains(t, err.Error(), "token") +} diff --git a/server/service/service_appconfig.go b/server/service/service_appconfig.go index c2d4ea477b..497203ca0e 100644 --- a/server/service/service_appconfig.go +++ b/server/service/service_appconfig.go @@ -235,6 +235,16 @@ func (svc *Service) LoggingConfig(ctx context.Context) (*fleet.Logging, error) { Server: conf.Nats.Server, }, } + case "splunk": + *lp.target = fleet.LoggingPlugin{ + Plugin: "splunk", + Config: fleet.SplunkConfig{ + URL: conf.Splunk.URL, + Index: conf.Splunk.Index, + Source: conf.Splunk.Source, + SourceType: conf.Splunk.SourceType, + }, + } default: return nil, ctxerr.Errorf(ctx, "unrecognized logging plugin: %s", lp.plugin) }