Remove dead code

transform/native_logs_csv.go

@@ -132,7 +132,9 @@ func (w *NativeLogsCSVWriter) MarshalNativeLog(log *types.Log) error {
 	hasPrevField = false
 	for k, v := range log.Resource {
 		_, lifted := w.fieldLookup[k]
-		if strings.HasPrefix(k, "adxmon_") || lifted {
+
+		// These are added by collector and used internally.  Don't store them in the final table.
+		if strings.HasPrefix(k, "adxmon_") || strings.HasPrefix(k, "label.") || strings.HasPrefix(k, "annotation.") || lifted {
 			continue
 		}
 

transform/native_logs_csv_test.go

@@ -191,7 +191,7 @@ func TestMarshalCSV_NativeLog(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.Name, func(t *testing.T) {
 			var b bytes.Buffer
-			w := NewCSVWriter(&b, nil)
+			w := NewCSVNativeLogsCSVWriter(&b, nil)
 
 			for _, log := range tt.Batch.Logs {
 				err := w.MarshalNativeLog(log)
@@ -204,10 +204,10 @@ func TestMarshalCSV_NativeLog(t *testing.T) {
 			reader := csv.NewReader(&b)
 			records, err := reader.ReadAll()
 			require.NoError(t, err)
-			require.Len(t, records, len(tt.Expect))
+			require.Len(t, records, len(tt.Expect)+1) // +1 for schema
 
 			for i, expect := range tt.Expect {
-				record, err := getLogRecord(records[i])
+				record, err := getLogRecord(records[i+1]) // +1 for schema
 				require.NoError(t, err)
 				require.Equal(t, expect, record)
 			}
@@ -238,7 +238,7 @@ func BenchmarkMarshalCSV_NativeLog(b *testing.B) {
 	}
 
 	buf := bytes.NewBuffer(make([]byte, 0, 64*1024))
-	enc := NewCSVWriter(buf, nil)
+	enc := NewCSVNativeLogsCSVWriter(buf, nil)
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {

transform/logs_csv.go → transform/otlp_logs_csv.go

@@ -9,9 +9,7 @@ import (
 	"time"
 
 	commonv1 "buf.build/gen/go/opentelemetry/opentelemetry/protocolbuffers/go/opentelemetry/proto/common/v1"
-	"github.com/Azure/adx-mon/collector/logs/types"
 	"github.com/Azure/adx-mon/pkg/otlp"
-	"github.com/pquerna/ffjson/ffjson"
 	fflib "github.com/pquerna/ffjson/fflib/v1"
 )
 
@@ -120,117 +118,6 @@ func (w *CSVWriter) MarshalLog(logs *otlp.Logs) error {
 	return w.enc.Error()
 }
 
-func (w *CSVWriter) MarshalNativeLog(log *types.Log) error {
-	// There are 9 fields defined in an OTLP log schema
-	fields := make([]string, 0, 9)
-	// Convert log records to CSV
-	// see samples at https://opentelemetry.io/docs/specs/otel/protocol/file-exporter/#examples
-	// Reset fields
-	fields = fields[:0]
-	// Timestamp
-	fields = append(fields, otlpTSToUTC(int64(log.Timestamp)))
-	// ObservedTimestamp
-	if v := log.ObservedTimestamp; v > 0 {
-		// Some clients don't set this value.
-		fields = append(fields, otlpTSToUTC(int64(log.ObservedTimestamp)))
-	} else {
-		fields = append(fields, time.Now().UTC().Format(time.RFC3339Nano))
-	}
-	// TraceId - we don't have this
-	fields = append(fields, "")
-	// SpanId - we don't have this
-	fields = append(fields, "")
-	// SeverityText - we don't have this
-	fields = append(fields, "")
-	// SeverityNumber - we don't have this
-	fields = append(fields, "")
-	// Body
-	buf := w.buf
-	buf.Reset()
-	buf.WriteByte('{')
-	hasPrevField := false
-	for k, v := range log.Body {
-		val, err := ffjson.Marshal(v)
-		if err != nil {
-			continue
-		}
-
-		if hasPrevField {
-			buf.WriteByte(',')
-		} else {
-			hasPrevField = true
-		}
-		fflib.WriteJson(buf, []byte(k))
-		buf.WriteByte(':')
-		buf.Write(val) // Already marshalled into json. Don't escape it again.
-		ffjson.Pool(val)
-	}
-	buf.WriteByte('}')
-	fields = append(fields, buf.String())
-
-	// Resource
-	buf.Reset()
-	buf.WriteByte('{')
-	hasPrevField = false
-	for k, v := range log.Resource {
-		// These are added by collector and used internally.  Don't store them in the final table.
-		if strings.HasPrefix(k, "adxmon_") || strings.HasPrefix(k, "label.") || strings.HasPrefix(k, "annotation.") {
-			continue
-		}
-
-		val, err := ffjson.Marshal(v)
-		if err != nil {
-			continue
-		}
-
-		if hasPrevField {
-			buf.WriteByte(',')
-		} else {
-			hasPrevField = true
-		}
-		fflib.WriteJson(buf, []byte(k))
-		buf.WriteByte(':')
-		buf.Write(val) // Already marshalled into json. Don't escape it again.
-		ffjson.Pool(val)
-	}
-	buf.WriteByte('}')
-	fields = append(fields, buf.String())
-
-	// Attributes
-	buf.Reset()
-	buf.WriteByte('{')
-	hasPrevField = false
-	for k, v := range log.Attributes {
-		if strings.HasPrefix(k, "adxmon_") {
-			continue
-		}
-
-		val, err := ffjson.Marshal(v)
-		if err != nil {
-			continue
-		}
-
-		if hasPrevField {
-			buf.WriteByte(',')
-		} else {
-			hasPrevField = true
-		}
-		fflib.WriteJson(buf, []byte(k))
-		buf.WriteByte(':')
-		buf.Write(val) // Already marshalled into json. Don't escape it again.
-		ffjson.Pool(val)
-	}
-	buf.WriteByte('}')
-	fields = append(fields, buf.String())
-	// Serialize
-	if err := w.enc.Write(fields); err != nil {
-		return err
-	}
-
-	w.enc.Flush()
-	return w.enc.Error()
-}
-
 func (w *CSVWriter) Reset() {
 	w.w.Reset()
 	w.buf.Reset()