diff --git a/changes/issue-3911-kafka-rest-proxy-header-value b/changes/issue-3911-kafka-rest-proxy-header-value new file mode 100644 index 0000000000..621d079fb5 --- /dev/null +++ b/changes/issue-3911-kafka-rest-proxy-header-value @@ -0,0 +1 @@ +* Add the ability to specify the Content Type Value for Kafka REST proxy logging plugin, this defaults to "application/vnd.kafka.json.v1+json" to not break existing integrations diff --git a/docs/Deploying/Configuration.md b/docs/Deploying/Configuration.md index feb6df477c..1e2db92dd6 100644 --- a/docs/Deploying/Configuration.md +++ b/docs/Deploying/Configuration.md @@ -1641,7 +1641,7 @@ The URL of the host which to check for the topic existence and post messages to - Environment variable: `FLEET_KAFKAREST_PROXYHOST` - Config file format: - ``` + ```yaml kafkarest: proxyhost: "https://localhost:8443" ``` @@ -1656,7 +1656,7 @@ The identifier of the kafka topic that osquery status logs will be published to. - Environment variable: `FLEET_KAFKAREST_STATUS_TOPIC` - Config file format: - ``` + ```yaml kafkarest: status_topic: osquery_status ``` @@ -1671,7 +1671,7 @@ The identifier of the kafka topic that osquery result logs will be published to. - Environment variable: `FLEET_KAFKAREST_RESULT_TOPIC` - Config file format: - ``` + ```yaml kafkarest: status_topic: osquery_result ``` @@ -1686,11 +1686,28 @@ The timeout value for the http post attempt. Value is in units of seconds. - Environment variable: `FLEET_KAFKAREST_TIMEOUT` - Config file format: - ``` + ```yaml kafkarest: timeout: 5 ``` +##### kafkarest_content_type_value + +This flag only has effect if `osquery_status_log_plugin` is set to `kafkarest`. + +The value of the Content-Type header to use in Kafka REST Proxy API calls. More information about available versions +can be found [here](https://docs.confluent.io/platform/current/kafka-rest/api.html#content-types). _Note: only JSON format is supported_ + +- Default value: application/vnd.kafka.json.v1+json +- Environment variable: `FLEET_KAFKAREST_CONTENT_TYPE_VALUE` +- Config file format: + + ```yaml + kafkarest: + content_type_value: application/vnd.kafka.json.v2+json + ``` + + #### S3 file carving backend ##### s3_bucket diff --git a/server/config/config.go b/server/config/config.go index a00b6c0caf..79f717f681 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -202,10 +202,11 @@ type FilesystemConfig struct { // KafkaRESTConfig defines configs for the Kafka REST Proxy logging plugin. type KafkaRESTConfig struct { - StatusTopic string `json:"status_topic" yaml:"status_topic"` - ResultTopic string `json:"result_topic" yaml:"result_topic"` - ProxyHost string `json:"proxyhost" yaml:"proxyhost"` - Timeout int `json:"timeout" yaml:"timeout"` + StatusTopic string `json:"status_topic" yaml:"status_topic"` + ResultTopic string `json:"result_topic" yaml:"result_topic"` + ProxyHost string `json:"proxyhost" yaml:"proxyhost"` + ContentTypeValue string `json:"content_type_value" yaml:"content_type_value"` + Timeout int `json:"timeout" yaml:"timeout"` } // LicenseConfig defines configs related to licensing Fleet. @@ -526,6 +527,8 @@ func (man Manager) addConfigs() { man.addConfigString("kafkarest.status_topic", "", "Kafka REST topic for status logs") man.addConfigString("kafkarest.result_topic", "", "Kafka REST topic for result logs") man.addConfigString("kafkarest.proxyhost", "", "Kafka REST proxy host url") + man.addConfigString("kafkarest.content_type_value", "application/vnd.kafka.json.v1+json", + "Kafka REST proxy content type header (defaults to \"application/vnd.kafka.json.v1+json\"") man.addConfigInt("kafkarest.timeout", 5, "Kafka REST proxy json post timeout") // License @@ -708,10 +711,11 @@ func (man Manager) LoadConfig() FleetConfig { EnableLogCompression: man.getConfigBool("filesystem.enable_log_compression"), }, KafkaREST: KafkaRESTConfig{ - StatusTopic: man.getConfigString("kafkarest.status_topic"), - ResultTopic: man.getConfigString("kafkarest.result_topic"), - ProxyHost: man.getConfigString("kafkarest.proxyhost"), - Timeout: man.getConfigInt("kafkarest.timeout"), + StatusTopic: man.getConfigString("kafkarest.status_topic"), + ResultTopic: man.getConfigString("kafkarest.result_topic"), + ProxyHost: man.getConfigString("kafkarest.proxyhost"), + ContentTypeValue: man.getConfigString("kafkarest.content_type_value"), + Timeout: man.getConfigInt("kafkarest.timeout"), }, License: LicenseConfig{ Key: man.getConfigString("license.key"), diff --git a/server/logging/firehose_test.go b/server/logging/firehose_test.go index 229b29e755..531acc332c 100644 --- a/server/logging/firehose_test.go +++ b/server/logging/firehose_test.go @@ -17,14 +17,14 @@ import ( var ( logs = []json.RawMessage{ - json.RawMessage(`{"foo": "bar"}`), - json.RawMessage(`{"flim": "flam"}`), - json.RawMessage(`{"jim": "jom"}`), + json.RawMessage(`{"foo":"bar"}`), + json.RawMessage(`{"flim":"flam"}`), + json.RawMessage(`{"jim":"jom"}`), } logsWithNewlines = []json.RawMessage{ - json.RawMessage(`{"foo": "bar"}` + "\n"), - json.RawMessage(`{"flim": "flam"}` + "\n"), - json.RawMessage(`{"jim": "jom"}` + "\n"), + json.RawMessage(`{"foo":"bar"}` + "\n"), + json.RawMessage(`{"flim":"flam"}` + "\n"), + json.RawMessage(`{"jim":"jom"}` + "\n"), } ) diff --git a/server/logging/kafkarest.go b/server/logging/kafkarest.go index 36cfffd6f3..1ecc548eb9 100644 --- a/server/logging/kafkarest.go +++ b/server/logging/kafkarest.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "io" "io/ioutil" "net/http" "time" @@ -14,7 +15,6 @@ import ( ) const ( - krContentTypeValue = "application/vnd.kafka.json.v1+json" krContentTypeHeader = "Content-Type" krTimestampHeader = "TimeStamp" krPublishTopicURL = "%s/topics/%s" @@ -22,15 +22,17 @@ const ( ) type KafkaRESTParams struct { - KafkaProxyHost string - KafkaTopic string - KafkaTimeout int + KafkaProxyHost string + KafkaTopic string + KafkaContentTypeValue string + KafkaTimeout int } type kafkaRESTProducer struct { - client *http.Client - URL string - CheckURL string + client *http.Client + URL string + CheckURL string + ContentTypeValue string } type kafkaRecords struct { @@ -43,9 +45,10 @@ type kafkaValue struct { func NewKafkaRESTWriter(p *KafkaRESTParams) (*kafkaRESTProducer, error) { producer := &kafkaRESTProducer{ - URL: fmt.Sprintf(krPublishTopicURL, p.KafkaProxyHost, p.KafkaTopic), - CheckURL: fmt.Sprintf(krCheckTopicURL, p.KafkaProxyHost, p.KafkaTopic), - client: fleethttp.NewClient(fleethttp.WithTimeout(time.Duration(p.KafkaTimeout) * time.Second)), + URL: fmt.Sprintf(krPublishTopicURL, p.KafkaProxyHost, p.KafkaTopic), + CheckURL: fmt.Sprintf(krCheckTopicURL, p.KafkaProxyHost, p.KafkaTopic), + client: fleethttp.NewClient(fleethttp.WithTimeout(time.Duration(p.KafkaTimeout) * time.Second)), + ContentTypeValue: p.KafkaContentTypeValue, } return producer, producer.checkTopic() @@ -95,14 +98,14 @@ func (l *kafkaRESTProducer) checkTopic() (err error) { return checkResponse(resp) } -func (l *kafkaRESTProducer) post(url string, buf *bytes.Buffer) (*http.Response, error) { +func (l *kafkaRESTProducer) post(url string, buf io.Reader) (*http.Response, error) { req, err := http.NewRequest(http.MethodPost, url, buf) if err != nil { return nil, fmt.Errorf("kafka rest new request: %w", err) } now := float64(time.Now().UnixNano()) / float64(time.Second) - req.Header.Set(krContentTypeHeader, krContentTypeValue) + req.Header.Set(krContentTypeHeader, l.ContentTypeValue) req.Header.Set(krTimestampHeader, fmt.Sprintf("%f", now)) return l.client.Do(req) diff --git a/server/logging/kafkarest_test.go b/server/logging/kafkarest_test.go new file mode 100644 index 0000000000..1c9f929ff8 --- /dev/null +++ b/server/logging/kafkarest_test.go @@ -0,0 +1,56 @@ +package logging + +import ( + "context" + "encoding/json" + "fmt" + "github.com/stretchr/testify/require" + "io/ioutil" + "net/http" + "net/http/httptest" + "testing" +) + +func TestKafkaRestWrite(t *testing.T) { + ctx := context.Background() + + var buf []byte + var err error + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + buf, err = ioutil.ReadAll(r.Body) + require.NoError(t, err) + require.Equal(t, r.URL.Path, "/topics/foo") + require.Equal(t, r.Header.Get("Content-Type"), "foobar") + w.WriteHeader(200) + })) + defer server.Close() + + producer := &kafkaRESTProducer{ + client: server.Client(), + URL: fmt.Sprintf(krPublishTopicURL, server.URL, "foo"), + CheckURL: fmt.Sprintf(krCheckTopicURL, server.URL, "foo"), + ContentTypeValue: "foobar", + } + + err = producer.Write(ctx, logs) + require.NoError(t, err) + + expected := makeKafkaRecords(logs) + var actual kafkaRecords + err = json.Unmarshal(buf, &actual) + require.NoError(t, err) + require.Equal(t, expected, actual) +} + +func makeKafkaRecords(messages []json.RawMessage) kafkaRecords { + data := kafkaRecords{ + Records: make([]kafkaValue, len(messages)), + } + + for i, log := range messages { + data.Records[i] = kafkaValue{ + Value: log, + } + } + return data +} diff --git a/server/logging/logging.go b/server/logging/logging.go index 68d4e4fb09..481f3ff60f 100644 --- a/server/logging/logging.go +++ b/server/logging/logging.go @@ -90,9 +90,10 @@ func New(config config.FleetConfig, logger log.Logger) (*OsqueryLogger, error) { } case "kafkarest": status, err = NewKafkaRESTWriter(&KafkaRESTParams{ - KafkaProxyHost: config.KafkaREST.ProxyHost, - KafkaTopic: config.KafkaREST.StatusTopic, - KafkaTimeout: config.KafkaREST.Timeout, + KafkaProxyHost: config.KafkaREST.ProxyHost, + KafkaTopic: config.KafkaREST.StatusTopic, + KafkaContentTypeValue: config.KafkaREST.ContentTypeValue, + KafkaTimeout: config.KafkaREST.Timeout, }) if err != nil { return nil, fmt.Errorf("create kafka rest status logger: %w", err) @@ -173,9 +174,10 @@ func New(config config.FleetConfig, logger log.Logger) (*OsqueryLogger, error) { } case "kafkarest": result, err = NewKafkaRESTWriter(&KafkaRESTParams{ - KafkaProxyHost: config.KafkaREST.ProxyHost, - KafkaTopic: config.KafkaREST.ResultTopic, - KafkaTimeout: config.KafkaREST.Timeout, + KafkaProxyHost: config.KafkaREST.ProxyHost, + KafkaTopic: config.KafkaREST.ResultTopic, + KafkaContentTypeValue: config.KafkaREST.ContentTypeValue, + KafkaTimeout: config.KafkaREST.Timeout, }) if err != nil { return nil, fmt.Errorf("create kafka rest result logger: %w", err)