Merge #77210

77210: util/log: only compute content-type once r=rimadeodhar a=knz

This is a fixup to #77014:

- the content-type is not a property of the format *parser*, it's a
  property of the format itself. Touching `formatParsers` to compute
  it is an abstraction mismatch.

- the content-type is constant for a given log config. There's no
  need to re-compute it on every log entry. It can be computed
  and stored once when the config is applied in `flags.go`.

  (Also, the `outputLogEntry` code is time-sensitive, so we don't want
  to give it more work.)

- the content-type is only needed for HTTP sinks. No need
  to compute it for other sink types. So its retrieval can be
  confined to `newHTTPSink`.

Release justification: bug fix

Release note: None

Co-authored-by: Raphael 'kena' Poss <knz@thaumogen.net>

pkg/util/log/clog.go

@@ -333,8 +333,7 @@ func (l *loggerT) outputLogEntry(entry logEntry) {
 				// The sink was not accepting entries at this level. Nothing to do.
 				continue
 			}
-			format := formatParsers[s.formatter.formatterName()]
-			if err := s.sink.output(bufs.b[i].Bytes(), sinkOutputOptions{extraFlush: extraFlush, forceSync: isFatal, format: format}); err != nil {
+			if err := s.sink.output(bufs.b[i].Bytes(), sinkOutputOptions{extraFlush: extraFlush, forceSync: isFatal}); err != nil {
 				if !s.criticality {
 					// An error on this sink is not critical. Just report
 					// the error and move on.

pkg/util/log/flags.go

@@ -385,6 +385,7 @@ func newHTTPSinkInfo(c logconfig.HTTPSinkConfig) (*sinkInfo, error) {
 		unsafeTLS:         *c.UnsafeTLS,
 		timeout:           *c.Timeout,
 		disableKeepAlives: *c.DisableKeepAlives,
+		contentType:       info.formatter.contentType(),
 	})
 	if err != nil {
 		return nil, err

pkg/util/log/format_crdb_v1.go

@@ -37,6 +37,8 @@ func (formatCrdbV1) formatEntry(entry logEntry) *buffer {
 
 func (formatCrdbV1) doc() string { return formatCrdbV1CommonDoc(false /* withCounter */) }
 
+func (formatCrdbV1) contentType() string { return "text/plain" }
+
 func formatCrdbV1CommonDoc(withCounter bool) string {
 	var buf strings.Builder
 
@@ -167,6 +169,8 @@ func (formatCrdbV1WithCounter) formatEntry(entry logEntry) *buffer {
 
 func (formatCrdbV1WithCounter) doc() string { return formatCrdbV1CommonDoc(true /* withCounter */) }
 
+func (formatCrdbV1WithCounter) contentType() string { return "text/plain" }
+
 // formatCrdbV1TTY is like formatCrdbV1 and includes VT color codes if
 // the stderr output is a TTY and -nocolor is not passed on the
 // command line.
@@ -192,6 +196,8 @@ func (formatCrdbV1TTY) doc() string {
 	return "Same textual format as `" + formatCrdbV1{}.formatterName() + "`." + ttyFormatDoc
 }
 
+func (formatCrdbV1TTY) contentType() string { return "text/plain" }
+
 // formatCrdbV1ColorWithCounter is like formatCrdbV1WithCounter and
 // includes VT color codes if the stderr output is a TTY and -nocolor
 // is not passed on the command line.
@@ -211,6 +217,8 @@ func (formatCrdbV1TTYWithCounter) doc() string {
 	return "Same textual format as `" + formatCrdbV1WithCounter{}.formatterName() + "`." + ttyFormatDoc
 }
 
+func (formatCrdbV1TTYWithCounter) contentType() string { return "text/plain" }
+
 // formatEntryInternalV1 renders a log entry.
 // Log lines are colorized depending on severity.
 // It uses a newly allocated *buffer. The caller is responsible

pkg/util/log/format_crdb_v2.go

@@ -40,6 +40,8 @@ func (formatCrdbV2) formatEntry(entry logEntry) *buffer {
 
 func (formatCrdbV2) doc() string { return formatCrdbV2CommonDoc() }
 
+func (formatCrdbV2) contentType() string { return "text/plain" }
+
 func formatCrdbV2CommonDoc() string {
 	var buf strings.Builder
 
@@ -188,6 +190,8 @@ func (formatCrdbV2TTY) doc() string {
 	return "Same textual format as `" + formatCrdbV2{}.formatterName() + "`." + ttyFormatDoc
 }
 
+func (formatCrdbV2TTY) contentType() string { return "text/plain" }
+
 // formatEntryInternalV2 renders a log entry.
 // Log lines are colorized depending on severity.
 // It uses a newly allocated *buffer. The caller is responsible

pkg/util/log/format_json.go

@@ -31,6 +31,8 @@ func (f formatFluentJSONCompact) formatEntry(entry logEntry) *buffer {
 	return formatJSON(entry, true /* fluent */, tagCompact)
 }
 
+func (formatFluentJSONCompact) contentType() string { return "application/json" }
+
 type formatFluentJSONFull struct{}
 
 func (formatFluentJSONFull) formatterName() string { return "json-fluent" }
@@ -41,6 +43,8 @@ func (f formatFluentJSONFull) formatEntry(entry logEntry) *buffer {
 
 func (formatFluentJSONFull) doc() string { return formatJSONDoc(true /* fluent */, tagVerbose) }
 
+func (formatFluentJSONFull) contentType() string { return "application/json" }
+
 type formatJSONCompact struct{}
 
 func (formatJSONCompact) formatterName() string { return "json-compact" }
@@ -51,6 +55,8 @@ func (f formatJSONCompact) formatEntry(entry logEntry) *buffer {
 
 func (formatJSONCompact) doc() string { return formatJSONDoc(false /* fluent */, tagCompact) }
 
+func (formatJSONCompact) contentType() string { return "application/json" }
+
 type formatJSONFull struct{}
 
 func (formatJSONFull) formatterName() string { return "json" }
@@ -61,6 +67,8 @@ func (f formatJSONFull) formatEntry(entry logEntry) *buffer {
 
 func (formatJSONFull) doc() string { return formatJSONDoc(false /* fluent */, tagVerbose) }
 
+func (formatJSONFull) contentType() string { return "application/json" }
+
 func formatJSONDoc(forFluent bool, tags tagChoice) string {
 	var buf strings.Builder
 	buf.WriteString(`This format emits log entries as a JSON payload.

pkg/util/log/formats.go

@@ -17,6 +17,10 @@ type logFormatter interface {
 	// formatEntry formats a logEntry into a newly allocated *buffer.
 	// The caller is responsible for calling putBuffer() afterwards.
 	formatEntry(entry logEntry) *buffer
+
+	// contentType is the MIME content-type field to use on
+	// transports which use this metadata.
+	contentType() string
 }
 
 var formatParsers = map[string]string{

pkg/util/log/http_sink.go

@@ -30,15 +30,7 @@ type httpSinkOptions struct {
 	timeout           time.Duration
 	method            string
 	disableKeepAlives bool
-}
-
-// formatToContentType map contains a mapping from the log format
-// to the content type header that should be set for that format
-// for the HTTP POST method. The content type header defaults
-// to `text/plain` when the http sink is configured with a format
-// not included in this map.
-var formatToContentType = map[string]string{
-	"json": "application/json",
+	contentType       string
 }
 
 func newHTTPSink(url string, opt httpSinkOptions) (*httpSink, error) {
@@ -53,8 +45,9 @@ func newHTTPSink(url string, opt httpSinkOptions) (*httpSink, error) {
 			Transport: transport,
 			Timeout:   opt.timeout,
 		},
-		address:   url,
-		doRequest: doPost,
+		address:     url,
+		doRequest:   doPost,
+		contentType: "application/octet-stream",
 	}
 
 	if opt.unsafeTLS {
@@ -65,13 +58,18 @@ func newHTTPSink(url string, opt httpSinkOptions) (*httpSink, error) {
 		hs.doRequest = doGet
 	}
 
+	if opt.contentType != "" {
+		hs.contentType = opt.contentType
+	}
+
 	return hs, nil
 }
 
 type httpSink struct {
-	client    http.Client
-	address   string
-	doRequest func(sink *httpSink, logEntry []byte, format string) (*http.Response, error)
+	client      http.Client
+	address     string
+	contentType string
+	doRequest   func(sink *httpSink, logEntry []byte) (*http.Response, error)
 }
 
 // output emits some formatted bytes to this sink.
@@ -83,33 +81,30 @@ type httpSink struct {
 // sinks must not recursively call into logging when implementing
 // this method.
 func (hs *httpSink) output(b []byte, opt sinkOutputOptions) (err error) {
-	resp, err := hs.doRequest(hs, b, opt.format)
+	resp, err := hs.doRequest(hs, b)
 	if err != nil {
 		return err
 	}
 
 	if resp.StatusCode >= 400 {
 		return HTTPLogError{
 			StatusCode: resp.StatusCode,
-			Address:    hs.address}
+			Address:    hs.address,
+		}
 	}
 	return nil
 }
 
-func doPost(hs *httpSink, b []byte, format string) (*http.Response, error) {
-	contentType, ok := formatToContentType[format]
-	if !ok {
-		contentType = "text/plain"
-	}
-	resp, err := hs.client.Post(hs.address, contentType, bytes.NewReader(b))
+func doPost(hs *httpSink, b []byte) (*http.Response, error) {
+	resp, err := hs.client.Post(hs.address, hs.contentType, bytes.NewReader(b))
 	if err != nil {
 		return nil, err
 	}
 	resp.Body.Close() // don't care about content
 	return resp, nil
 }
 
-func doGet(hs *httpSink, b []byte, _ string) (*http.Response, error) {
+func doGet(hs *httpSink, b []byte) (*http.Response, error) {
 	resp, err := hs.client.Get(hs.address + "?" + url.QueryEscape(string(b)))
 	if err != nil {
 		return nil, err

pkg/util/log/http_sink_test.go

@@ -214,7 +214,7 @@ func TestHTTPSinkContentTypeJSON(t *testing.T) {
 	address := "http://localhost" // testBase appends the port
 	timeout := 5 * time.Second
 	tb := true
-	format := "json"
+	format := "json-fluent"
 	expectedContentType := "application/json"
 	defaults := logconfig.HTTPDefaults{
 		Address: &address,
@@ -232,7 +232,7 @@ func TestHTTPSinkContentTypeJSON(t *testing.T) {
 		t.Log(body)
 		contentType := header.Get("Content-Type")
 		if contentType != expectedContentType {
-			return errors.Newf("mismatched content type: expected %s, got %s")
+			return errors.Newf("mismatched content type: expected %s, got %s", expectedContentType, contentType)
 		}
 		return nil
 	}
@@ -266,7 +266,7 @@ func TestHTTPSinkContentTypePlainText(t *testing.T) {
 		t.Log(body)
 		contentType := header.Get("Content-Type")
 		if contentType != expectedContentType {
-			return errors.Newf("mismatched content type: expected %s, got %s")
+			return errors.Newf("mismatched content type: expected %s, got %s", expectedContentType, contentType)
 		}
 		return nil
 	}

pkg/util/log/intercept.go

@@ -73,6 +73,7 @@ type formatInterceptor struct{}
 
 func (formatInterceptor) formatterName() string { return "json-intercept" }
 func (formatInterceptor) doc() string           { return "internal only" }
+func (formatInterceptor) contentType() string   { return "application/json" }
 func (formatInterceptor) formatEntry(entry logEntry) *buffer {
 	pEntry := entry.convertToLegacy()
 	buf := getBuffer()

pkg/util/log/sinks.go

@@ -31,8 +31,6 @@ type sinkOutputOptions struct {
 	// forceSync forces synchronous operation of this output operation.
 	// That is, it will block until the output has been handled.
 	forceSync bool
-	// format specifies the log entry format (e.g. json etc.).
-	format string
 }
 
 // logSink abstracts the destination of logging events, after all