package splunk // import "github.com/docker/docker/daemon/logger/splunk" import ( "compress/gzip" "context" "fmt" "net/http" "os" "runtime" "testing" "time" "github.com/docker/docker/daemon/logger" "gotest.tools/v3/assert" "gotest.tools/v3/env" ) // Validate options func TestValidateLogOpt(t *testing.T) { err := ValidateLogOpt(map[string]string{ splunkURLKey: "http://127.0.0.1", splunkTokenKey: "2160C7EF-2CE9-4307-A180-F852B99CF417", splunkSourceKey: "mysource", splunkSourceTypeKey: "mysourcetype", splunkIndexKey: "myindex", splunkCAPathKey: "/usr/cert.pem", splunkCANameKey: "ca_name", splunkInsecureSkipVerifyKey: "true", splunkFormatKey: "json", splunkVerifyConnectionKey: "true", splunkGzipCompressionKey: "true", splunkGzipCompressionLevelKey: "1", envKey: "a", envRegexKey: "^foo", labelsKey: "b", tagKey: "c", }) if err != nil { t.Fatal(err) } err = ValidateLogOpt(map[string]string{ "not-supported-option": "a", }) if err == nil { t.Fatal("Expecting error on unsupported options") } } // Driver require user to specify required options func TestNewMissedConfig(t *testing.T) { info := logger.Info{ Config: map[string]string{}, } _, err := New(info) if err == nil { t.Fatal("Logger driver should fail when no required parameters specified") } } // Driver require user to specify splunk-url func TestNewMissedUrl(t *testing.T) { info := logger.Info{ Config: map[string]string{ splunkTokenKey: "4642492F-D8BD-47F1-A005-0C08AE4657DF", }, } _, err := New(info) if err.Error() != "splunk: splunk-url is expected" { t.Fatal("Logger driver should fail when no required parameters specified") } } // Driver require user to specify splunk-token func TestNewMissedToken(t *testing.T) { info := logger.Info{ Config: map[string]string{ splunkURLKey: "http://127.0.0.1:8088", }, } _, err := New(info) if err.Error() != "splunk: splunk-token is expected" { t.Fatal("Logger driver should fail when no required parameters specified") } } func TestNewWithProxy(t *testing.T) { proxy := "http://proxy.testing:8888" reset := env.Patch(t, "HTTP_PROXY", proxy) defer reset() // must not be localhost splunkURL := "http://example.com:12345" logger, err := New(logger.Info{ Config: map[string]string{ splunkURLKey: splunkURL, splunkTokenKey: "token", splunkVerifyConnectionKey: "false", }, ContainerID: "containeriid", }) assert.NilError(t, err) splunkLogger := logger.(*splunkLoggerInline) proxyFunc := splunkLogger.transport.Proxy assert.Assert(t, proxyFunc != nil) req, err := http.NewRequest(http.MethodGet, splunkURL, nil) assert.NilError(t, err) proxyURL, err := proxyFunc(req) assert.NilError(t, err) assert.Assert(t, proxyURL != nil) assert.Equal(t, proxy, proxyURL.String()) } // Test default settings func TestDefault(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, }, ContainerID: "containeriid", ContainerName: "container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } hostname, err := info.Hostname() if err != nil { t.Fatal(err) } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if loggerDriver.Name() != driverName { t.Fatal("Unexpected logger driver name") } if !hec.connectionVerified { t.Fatal("By default connection should be verified") } splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerInline) if !ok { t.Fatal("Unexpected Splunk Logging Driver type") } if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" || splunkLoggerDriver.auth != "Splunk "+hec.token || splunkLoggerDriver.nullMessage.Host != hostname || splunkLoggerDriver.nullMessage.Source != "" || splunkLoggerDriver.nullMessage.SourceType != "" || splunkLoggerDriver.nullMessage.Index != "" || splunkLoggerDriver.gzipCompression || splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency || splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize || splunkLoggerDriver.bufferMaximum != defaultBufferMaximum || cap(splunkLoggerDriver.stream) != defaultStreamChannelSize { t.Fatal("Found not default values setup in Splunk Logging Driver.") } message1Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("{\"a\":\"b\"}"), Source: "stdout", Timestamp: message1Time}); err != nil { t.Fatal(err) } message2Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("notajson"), Source: "stdout", Timestamp: message2Time}); err != nil { t.Fatal(err) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 2 { t.Fatal("Expected two messages") } if *hec.gzipEnabled { t.Fatal("Gzip should not be used") } message1 := hec.messages[0] if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) || message1.Host != hostname || message1.Source != "" || message1.SourceType != "" || message1.Index != "" { t.Fatalf("Unexpected values of message 1 %v", message1) } if event, err := message1.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != "{\"a\":\"b\"}" || event["source"] != "stdout" || event["tag"] != "containeriid" || len(event) != 3 { t.Fatalf("Unexpected event in message %v", event) } } message2 := hec.messages[1] if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) || message2.Host != hostname || message2.Source != "" || message2.SourceType != "" || message2.Index != "" { t.Fatalf("Unexpected values of message 1 %v", message2) } if event, err := message2.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != "notajson" || event["source"] != "stdout" || event["tag"] != "containeriid" || len(event) != 3 { t.Fatalf("Unexpected event in message %v", event) } } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify inline format with a not default settings for most of options func TestInlineFormatWithNonDefaultOptions(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkSourceKey: "mysource", splunkSourceTypeKey: "mysourcetype", splunkIndexKey: "myindex", splunkFormatKey: splunkFormatInline, splunkGzipCompressionKey: "true", tagKey: "{{.ImageName}}/{{.Name}}", labelsKey: "a", envRegexKey: "^foo", }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", ContainerLabels: map[string]string{ "a": "b", }, ContainerEnv: []string{"foo_finder=bar"}, } hostname, err := info.Hostname() if err != nil { t.Fatal(err) } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if !hec.connectionVerified { t.Fatal("By default connection should be verified") } splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerInline) if !ok { t.Fatal("Unexpected Splunk Logging Driver type") } if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" || splunkLoggerDriver.auth != "Splunk "+hec.token || splunkLoggerDriver.nullMessage.Host != hostname || splunkLoggerDriver.nullMessage.Source != "mysource" || splunkLoggerDriver.nullMessage.SourceType != "mysourcetype" || splunkLoggerDriver.nullMessage.Index != "myindex" || !splunkLoggerDriver.gzipCompression || splunkLoggerDriver.gzipCompressionLevel != gzip.DefaultCompression || splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency || splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize || splunkLoggerDriver.bufferMaximum != defaultBufferMaximum || cap(splunkLoggerDriver.stream) != defaultStreamChannelSize { t.Fatal("Values do not match configuration.") } messageTime := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("1"), Source: "stdout", Timestamp: messageTime}); err != nil { t.Fatal(err) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 1 { t.Fatal("Expected one message") } if !*hec.gzipEnabled { t.Fatal("Gzip should be used") } message := hec.messages[0] if message.Time != fmt.Sprintf("%f", float64(messageTime.UnixNano())/float64(time.Second)) || message.Host != hostname || message.Source != "mysource" || message.SourceType != "mysourcetype" || message.Index != "myindex" { t.Fatalf("Unexpected values of message %v", message) } if event, err := message.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != "1" || event["source"] != "stdout" || event["tag"] != "container_image_name/container_name" || event["attrs"].(map[string]interface{})["a"] != "b" || event["attrs"].(map[string]interface{})["foo_finder"] != "bar" || len(event) != 4 { t.Fatalf("Unexpected event in message %v", event) } } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify JSON format func TestJsonFormat(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkFormatKey: splunkFormatJSON, splunkGzipCompressionKey: "true", splunkGzipCompressionLevelKey: "1", }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } hostname, err := info.Hostname() if err != nil { t.Fatal(err) } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if !hec.connectionVerified { t.Fatal("By default connection should be verified") } splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerJSON) if !ok { t.Fatal("Unexpected Splunk Logging Driver type") } if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" || splunkLoggerDriver.auth != "Splunk "+hec.token || splunkLoggerDriver.nullMessage.Host != hostname || splunkLoggerDriver.nullMessage.Source != "" || splunkLoggerDriver.nullMessage.SourceType != "" || splunkLoggerDriver.nullMessage.Index != "" || !splunkLoggerDriver.gzipCompression || splunkLoggerDriver.gzipCompressionLevel != gzip.BestSpeed || splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency || splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize || splunkLoggerDriver.bufferMaximum != defaultBufferMaximum || cap(splunkLoggerDriver.stream) != defaultStreamChannelSize { t.Fatal("Values do not match configuration.") } message1Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("{\"a\":\"b\"}"), Source: "stdout", Timestamp: message1Time}); err != nil { t.Fatal(err) } message2Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("notjson"), Source: "stdout", Timestamp: message2Time}); err != nil { t.Fatal(err) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 2 { t.Fatal("Expected two messages") } message1 := hec.messages[0] if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) || message1.Host != hostname || message1.Source != "" || message1.SourceType != "" || message1.Index != "" { t.Fatalf("Unexpected values of message 1 %v", message1) } if event, err := message1.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"].(map[string]interface{})["a"] != "b" || event["source"] != "stdout" || event["tag"] != "containeriid" || len(event) != 3 { t.Fatalf("Unexpected event in message 1 %v", event) } } message2 := hec.messages[1] if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) || message2.Host != hostname || message2.Source != "" || message2.SourceType != "" || message2.Index != "" { t.Fatalf("Unexpected values of message 2 %v", message2) } // If message cannot be parsed as JSON - it should be sent as a line if event, err := message2.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != "notjson" || event["source"] != "stdout" || event["tag"] != "containeriid" || len(event) != 3 { t.Fatalf("Unexpected event in message 2 %v", event) } } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify raw format func TestRawFormat(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkFormatKey: splunkFormatRaw, }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } hostname, err := info.Hostname() assert.NilError(t, err) loggerDriver, err := New(info) assert.NilError(t, err) if !hec.connectionVerified { t.Fatal("By default connection should be verified") } splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerRaw) if !ok { t.Fatal("Unexpected Splunk Logging Driver type") } if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" || splunkLoggerDriver.auth != "Splunk "+hec.token || splunkLoggerDriver.nullMessage.Host != hostname || splunkLoggerDriver.nullMessage.Source != "" || splunkLoggerDriver.nullMessage.SourceType != "" || splunkLoggerDriver.nullMessage.Index != "" || splunkLoggerDriver.gzipCompression || splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency || splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize || splunkLoggerDriver.bufferMaximum != defaultBufferMaximum || cap(splunkLoggerDriver.stream) != defaultStreamChannelSize || string(splunkLoggerDriver.prefix) != "containeriid " { t.Fatal("Values do not match configuration.") } message1Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("{\"a\":\"b\"}"), Source: "stdout", Timestamp: message1Time}); err != nil { t.Fatal(err) } message2Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("notjson"), Source: "stdout", Timestamp: message2Time}); err != nil { t.Fatal(err) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 2 { t.Fatal("Expected two messages") } message1 := hec.messages[0] if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) || message1.Host != hostname || message1.Source != "" || message1.SourceType != "" || message1.Index != "" { t.Fatalf("Unexpected values of message 1 %v", message1) } if event, err := message1.EventAsString(); err != nil { t.Fatal(err) } else { if event != "containeriid {\"a\":\"b\"}" { t.Fatalf("Unexpected event in message 1 %v", event) } } message2 := hec.messages[1] if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) || message2.Host != hostname || message2.Source != "" || message2.SourceType != "" || message2.Index != "" { t.Fatalf("Unexpected values of message 2 %v", message2) } if event, err := message2.EventAsString(); err != nil { t.Fatal(err) } else { if event != "containeriid notjson" { t.Fatalf("Unexpected event in message 1 %v", event) } } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify raw format with labels func TestRawFormatWithLabels(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkFormatKey: splunkFormatRaw, labelsKey: "a", }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", ContainerLabels: map[string]string{ "a": "b", }, } hostname, err := info.Hostname() if err != nil { t.Fatal(err) } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if !hec.connectionVerified { t.Fatal("By default connection should be verified") } splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerRaw) if !ok { t.Fatal("Unexpected Splunk Logging Driver type") } if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" || splunkLoggerDriver.auth != "Splunk "+hec.token || splunkLoggerDriver.nullMessage.Host != hostname || splunkLoggerDriver.nullMessage.Source != "" || splunkLoggerDriver.nullMessage.SourceType != "" || splunkLoggerDriver.nullMessage.Index != "" || splunkLoggerDriver.gzipCompression || splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency || splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize || splunkLoggerDriver.bufferMaximum != defaultBufferMaximum || cap(splunkLoggerDriver.stream) != defaultStreamChannelSize || string(splunkLoggerDriver.prefix) != "containeriid a=b " { t.Fatal("Values do not match configuration.") } message1Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("{\"a\":\"b\"}"), Source: "stdout", Timestamp: message1Time}); err != nil { t.Fatal(err) } message2Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("notjson"), Source: "stdout", Timestamp: message2Time}); err != nil { t.Fatal(err) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 2 { t.Fatal("Expected two messages") } message1 := hec.messages[0] if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) || message1.Host != hostname || message1.Source != "" || message1.SourceType != "" || message1.Index != "" { t.Fatalf("Unexpected values of message 1 %v", message1) } if event, err := message1.EventAsString(); err != nil { t.Fatal(err) } else { if event != "containeriid a=b {\"a\":\"b\"}" { t.Fatalf("Unexpected event in message 1 %v", event) } } message2 := hec.messages[1] if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) || message2.Host != hostname || message2.Source != "" || message2.SourceType != "" || message2.Index != "" { t.Fatalf("Unexpected values of message 2 %v", message2) } if event, err := message2.EventAsString(); err != nil { t.Fatal(err) } else { if event != "containeriid a=b notjson" { t.Fatalf("Unexpected event in message 2 %v", event) } } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify that Splunk Logging Driver can accept tag="" which will allow to send raw messages // in the same way we get them in stdout/stderr func TestRawFormatWithoutTag(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkFormatKey: splunkFormatRaw, tagKey: "", }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } hostname, err := info.Hostname() if err != nil { t.Fatal(err) } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if !hec.connectionVerified { t.Fatal("By default connection should be verified") } splunkLoggerDriver, ok := loggerDriver.(*splunkLoggerRaw) if !ok { t.Fatal("Unexpected Splunk Logging Driver type") } if splunkLoggerDriver.url != hec.URL()+"/services/collector/event/1.0" || splunkLoggerDriver.auth != "Splunk "+hec.token || splunkLoggerDriver.nullMessage.Host != hostname || splunkLoggerDriver.nullMessage.Source != "" || splunkLoggerDriver.nullMessage.SourceType != "" || splunkLoggerDriver.nullMessage.Index != "" || splunkLoggerDriver.gzipCompression || splunkLoggerDriver.postMessagesFrequency != defaultPostMessagesFrequency || splunkLoggerDriver.postMessagesBatchSize != defaultPostMessagesBatchSize || splunkLoggerDriver.bufferMaximum != defaultBufferMaximum || cap(splunkLoggerDriver.stream) != defaultStreamChannelSize || string(splunkLoggerDriver.prefix) != "" { t.Log(string(splunkLoggerDriver.prefix) + "a") t.Fatal("Values do not match configuration.") } message1Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("{\"a\":\"b\"}"), Source: "stdout", Timestamp: message1Time}); err != nil { t.Fatal(err) } message2Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte("notjson"), Source: "stdout", Timestamp: message2Time}); err != nil { t.Fatal(err) } message3Time := time.Now() if err := loggerDriver.Log(&logger.Message{Line: []byte(" "), Source: "stdout", Timestamp: message3Time}); err != nil { t.Fatal(err) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } // message3 would have an empty or whitespace only string in the "event" field // both of which are not acceptable to HEC // thus here we must expect 2 messages, not 3 if len(hec.messages) != 2 { t.Fatal("Expected two messages") } message1 := hec.messages[0] if message1.Time != fmt.Sprintf("%f", float64(message1Time.UnixNano())/float64(time.Second)) || message1.Host != hostname || message1.Source != "" || message1.SourceType != "" || message1.Index != "" { t.Fatalf("Unexpected values of message 1 %v", message1) } if event, err := message1.EventAsString(); err != nil { t.Fatal(err) } else { if event != "{\"a\":\"b\"}" { t.Fatalf("Unexpected event in message 1 %v", event) } } message2 := hec.messages[1] if message2.Time != fmt.Sprintf("%f", float64(message2Time.UnixNano())/float64(time.Second)) || message2.Host != hostname || message2.Source != "" || message2.SourceType != "" || message2.Index != "" { t.Fatalf("Unexpected values of message 2 %v", message2) } if event, err := message2.EventAsString(); err != nil { t.Fatal(err) } else { if event != "notjson" { t.Fatalf("Unexpected event in message 2 %v", event) } } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify that we will send messages in batches with default batching parameters, // but change frequency to be sure that numOfRequests will match expected 17 requests func TestBatching(t *testing.T) { if err := os.Setenv(envVarPostMessagesFrequency, "10h"); err != nil { t.Fatal(err) } hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } for i := 0; i < defaultStreamChannelSize*4; i++ { if err := loggerDriver.Log(&logger.Message{Line: []byte(fmt.Sprintf("%d", i)), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != defaultStreamChannelSize*4 { t.Fatal("Not all messages delivered") } for i, message := range hec.messages { if event, err := message.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != fmt.Sprintf("%d", i) { t.Fatalf("Unexpected event in message %v", event) } } } // 1 to verify connection and 16 batches if hec.numOfRequests != 17 { t.Fatalf("Unexpected number of requests %d", hec.numOfRequests) } err = hec.Close() if err != nil { t.Fatal(err) } if err := os.Setenv(envVarPostMessagesFrequency, ""); err != nil { t.Fatal(err) } } // Verify that test is using time to fire events not rare than specified frequency func TestFrequency(t *testing.T) { if err := os.Setenv(envVarPostMessagesFrequency, "5ms"); err != nil { t.Fatal(err) } hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } for i := 0; i < 10; i++ { if err := loggerDriver.Log(&logger.Message{Line: []byte(fmt.Sprintf("%d", i)), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } time.Sleep(15 * time.Millisecond) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 10 { t.Fatal("Not all messages delivered") } for i, message := range hec.messages { if event, err := message.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != fmt.Sprintf("%d", i) { t.Fatalf("Unexpected event in message %v", event) } } } // 1 to verify connection and 10 to verify that we have sent messages with required frequency, // but because frequency is too small (to keep test quick), instead of 11, use 9 if context switches will be slow expectedRequests := 9 if runtime.GOOS == "windows" { // sometimes in Windows, this test fails with number of requests showing 8. So be more conservative. expectedRequests = 7 } if hec.numOfRequests < expectedRequests { t.Fatalf("Unexpected number of requests %d", hec.numOfRequests) } err = hec.Close() if err != nil { t.Fatal(err) } if err := os.Setenv(envVarPostMessagesFrequency, ""); err != nil { t.Fatal(err) } } // Simulate behavior similar to first version of Splunk Logging Driver, when we were sending one message // per request func TestOneMessagePerRequest(t *testing.T) { if err := os.Setenv(envVarPostMessagesFrequency, "10h"); err != nil { t.Fatal(err) } if err := os.Setenv(envVarPostMessagesBatchSize, "1"); err != nil { t.Fatal(err) } if err := os.Setenv(envVarBufferMaximum, "1"); err != nil { t.Fatal(err) } if err := os.Setenv(envVarStreamChannelSize, "0"); err != nil { t.Fatal(err) } hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } for i := 0; i < 10; i++ { if err := loggerDriver.Log(&logger.Message{Line: []byte(fmt.Sprintf("%d", i)), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 10 { t.Fatal("Not all messages delivered") } for i, message := range hec.messages { if event, err := message.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != fmt.Sprintf("%d", i) { t.Fatalf("Unexpected event in message %v", event) } } } // 1 to verify connection and 10 messages if hec.numOfRequests != 11 { t.Fatalf("Unexpected number of requests %d", hec.numOfRequests) } err = hec.Close() if err != nil { t.Fatal(err) } if err := os.Setenv(envVarPostMessagesFrequency, ""); err != nil { t.Fatal(err) } if err := os.Setenv(envVarPostMessagesBatchSize, ""); err != nil { t.Fatal(err) } if err := os.Setenv(envVarBufferMaximum, ""); err != nil { t.Fatal(err) } if err := os.Setenv(envVarStreamChannelSize, ""); err != nil { t.Fatal(err) } } // Driver should not be created when HEC is unresponsive func TestVerify(t *testing.T) { hec := NewHTTPEventCollectorMock(t) hec.simulateServerError = true go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } _, err := New(info) if err == nil { t.Fatal("Expecting driver to fail, when server is unresponsive") } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify that user can specify to skip verification that Splunk HEC is working. // Also in this test we verify retry logic. func TestSkipVerify(t *testing.T) { hec := NewHTTPEventCollectorMock(t) hec.simulateServerError = true go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkVerifyConnectionKey: "false", }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if hec.connectionVerified { t.Fatal("Connection should not be verified") } for i := 0; i < defaultStreamChannelSize*2; i++ { if err := loggerDriver.Log(&logger.Message{Line: []byte(fmt.Sprintf("%d", i)), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } } if len(hec.messages) != 0 { t.Fatal("No messages should be accepted at this point") } hec.simulateErr(false) for i := defaultStreamChannelSize * 2; i < defaultStreamChannelSize*4; i++ { if err := loggerDriver.Log(&logger.Message{Line: []byte(fmt.Sprintf("%d", i)), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != defaultStreamChannelSize*4 { t.Fatal("Not all messages delivered") } for i, message := range hec.messages { if event, err := message.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != fmt.Sprintf("%d", i) { t.Fatalf("Unexpected event in message %v", event) } } } err = hec.Close() if err != nil { t.Fatal(err) } } // Verify logic for when we filled whole buffer func TestBufferMaximum(t *testing.T) { if err := os.Setenv(envVarPostMessagesBatchSize, "2"); err != nil { t.Fatal(err) } if err := os.Setenv(envVarBufferMaximum, "10"); err != nil { t.Fatal(err) } if err := os.Setenv(envVarStreamChannelSize, "0"); err != nil { t.Fatal(err) } hec := NewHTTPEventCollectorMock(t) hec.simulateErr(true) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkVerifyConnectionKey: "false", }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if hec.connectionVerified { t.Fatal("Connection should not be verified") } for i := 0; i < 11; i++ { if err := loggerDriver.Log(&logger.Message{Line: []byte(fmt.Sprintf("%d", i)), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } } if len(hec.messages) != 0 { t.Fatal("No messages should be accepted at this point") } hec.simulateServerError = false err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 9 { t.Fatalf("Expected # of messages %d, got %d", 9, len(hec.messages)) } // First 1000 messages are written to daemon log when buffer was full for i, message := range hec.messages { if event, err := message.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != fmt.Sprintf("%d", i+2) { t.Fatalf("Unexpected event in message %v", event) } } } err = hec.Close() if err != nil { t.Fatal(err) } if err := os.Setenv(envVarPostMessagesBatchSize, ""); err != nil { t.Fatal(err) } if err := os.Setenv(envVarBufferMaximum, ""); err != nil { t.Fatal(err) } if err := os.Setenv(envVarStreamChannelSize, ""); err != nil { t.Fatal(err) } } // Verify that we are not blocking close when HEC is down for the whole time func TestServerAlwaysDown(t *testing.T) { if err := os.Setenv(envVarPostMessagesBatchSize, "2"); err != nil { t.Fatal(err) } if err := os.Setenv(envVarBufferMaximum, "4"); err != nil { t.Fatal(err) } if err := os.Setenv(envVarStreamChannelSize, "0"); err != nil { t.Fatal(err) } hec := NewHTTPEventCollectorMock(t) hec.simulateServerError = true go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, splunkVerifyConnectionKey: "false", }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if hec.connectionVerified { t.Fatal("Connection should not be verified") } for i := 0; i < 5; i++ { if err := loggerDriver.Log(&logger.Message{Line: []byte(fmt.Sprintf("%d", i)), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if len(hec.messages) != 0 { t.Fatal("No messages should be sent") } err = hec.Close() if err != nil { t.Fatal(err) } if err := os.Setenv(envVarPostMessagesBatchSize, ""); err != nil { t.Fatal(err) } if err := os.Setenv(envVarBufferMaximum, ""); err != nil { t.Fatal(err) } if err := os.Setenv(envVarStreamChannelSize, ""); err != nil { t.Fatal(err) } } // Cannot send messages after we close driver func TestCannotSendAfterClose(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } loggerDriver, err := New(info) if err != nil { t.Fatal(err) } if err := loggerDriver.Log(&logger.Message{Line: []byte("message1"), Source: "stdout", Timestamp: time.Now()}); err != nil { t.Fatal(err) } err = loggerDriver.Close() if err != nil { t.Fatal(err) } if err := loggerDriver.Log(&logger.Message{Line: []byte("message2"), Source: "stdout", Timestamp: time.Now()}); err == nil { t.Fatal("Driver should not allow to send messages after close") } if len(hec.messages) != 1 { t.Fatal("Only one message should be sent") } message := hec.messages[0] if event, err := message.EventAsMap(); err != nil { t.Fatal(err) } else { if event["line"] != "message1" { t.Fatalf("Unexpected event in message %v", event) } } err = hec.Close() if err != nil { t.Fatal(err) } } func TestDeadlockOnBlockedEndpoint(t *testing.T) { hec := NewHTTPEventCollectorMock(t) go hec.Serve() info := logger.Info{ Config: map[string]string{ splunkURLKey: hec.URL(), splunkTokenKey: hec.token, }, ContainerID: "containeriid", ContainerName: "/container_name", ContainerImageID: "contaimageid", ContainerImageName: "container_image_name", } l, err := New(info) if err != nil { t.Fatal(err) } ctx, unblock := context.WithCancel(context.Background()) hec.withBlock(ctx) defer unblock() batchSendTimeout = 1 * time.Second if err := l.Log(&logger.Message{}); err != nil { t.Fatal(err) } done := make(chan struct{}) go func() { l.Close() close(done) }() select { case <-time.After(60 * time.Second): buf := make([]byte, 1e6) buf = buf[:runtime.Stack(buf, true)] t.Logf("STACK DUMP: \n\n%s\n\n", string(buf)) t.Fatal("timeout waiting for close to finish") case <-done: } }