feat: track host log loss from collector to Kusto

collector/logs/engine/batch.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/Azure/adx-mon/collector/logs/types"
+	"github.com/Azure/adx-mon/pkg/wal"
 )
 
 type BatchConfig struct {
@@ -13,6 +14,7 @@ type BatchConfig struct {
 	InputQueue   <-chan *types.Log
 	OutputQueue  chan<- *types.LogBatch
 	AckGenerator func(log *types.Log) func()
+	SampleType   wal.SampleType
 }
 
 func BatchLogs(ctx context.Context, config BatchConfig) {
@@ -21,6 +23,7 @@ func BatchLogs(ctx context.Context, config BatchConfig) {
 
 	currentBatch := types.LogBatchPool.Get(1024).(*types.LogBatch)
 	currentBatch.Reset()
+	currentBatch.SampleType = config.SampleType
 	for {
 		select {
 		case <-ctx.Done():
@@ -34,13 +37,15 @@ func BatchLogs(ctx context.Context, config BatchConfig) {
 				flush(config, currentBatch)
 				currentBatch = types.LogBatchPool.Get(1024).(*types.LogBatch)
 				currentBatch.Reset()
+				currentBatch.SampleType = config.SampleType
 			}
 		case msg := <-config.InputQueue:
 			currentBatch.Logs = append(currentBatch.Logs, msg)
 			if len(currentBatch.Logs) >= config.MaxBatchSize {
 				flush(config, currentBatch)
 				currentBatch = types.LogBatchPool.Get(1024).(*types.LogBatch)
 				currentBatch.Reset()
+				currentBatch.SampleType = config.SampleType
 				ticker.Reset(config.MaxBatchWait)
 			}
 		}

collector/logs/sources/journal/journal_linux.go

@@ -10,6 +10,7 @@ import (
 	"github.com/Azure/adx-mon/collector/logs/engine"
 	"github.com/Azure/adx-mon/collector/logs/types"
 	"github.com/Azure/adx-mon/pkg/logger"
+	"github.com/Azure/adx-mon/pkg/wal"
 	"github.com/coreos/go-systemd/sdjournal"
 )
 
@@ -102,6 +103,7 @@ func (s *Source) Open(ctx context.Context) error {
 			InputQueue:   batchQueue,
 			OutputQueue:  outputQueue,
 			AckGenerator: ackGenerator,
+			SampleType:   wal.HostLogSampleType,
 		}
 		s.wg.Add(1)
 		go func() {

collector/logs/sources/kernel/kernel.go

@@ -14,6 +14,7 @@ import (
 	"github.com/Azure/adx-mon/collector/logs/engine"
 	"github.com/Azure/adx-mon/collector/logs/types"
 	"github.com/Azure/adx-mon/pkg/logger"
+	"github.com/Azure/adx-mon/pkg/wal"
 	"github.com/siderolabs/go-kmsg"
 )
 
@@ -117,6 +118,7 @@ func (s *KernelSource) Open(ctx context.Context) error {
 		InputQueue:   batchQueue,
 		OutputQueue:  outputQueue,
 		AckGenerator: s.ackGenerator,
+		SampleType:   wal.HostLogSampleType,
 	}
 
 	s.wg.Add(1)

collector/logs/sources/tail/tailer.go

@@ -12,6 +12,7 @@ import (
 	"github.com/Azure/adx-mon/collector/logs/transforms/parser"
 	"github.com/Azure/adx-mon/collector/logs/types"
 	"github.com/Azure/adx-mon/pkg/logger"
+	"github.com/Azure/adx-mon/pkg/wal"
 	"github.com/tenebris-tech/tail"
 )
 
@@ -91,6 +92,7 @@ func StartTailing(config TailerConfig) (*Tailer, error) {
 		InputQueue:   batchQueue,
 		OutputQueue:  outputQueue,
 		AckGenerator: config.AckGenerator,
+		SampleType:   wal.HostLogSampleType,
 	}
 
 	tailer.wg.Add(1)

collector/logs/types/logs.go

@@ -7,6 +7,7 @@ import (
 	"sync/atomic"
 
 	"github.com/Azure/adx-mon/pkg/logger"
+	"github.com/Azure/adx-mon/pkg/wal"
 )
 
 var assertionsEnabledValue bool = false
@@ -243,8 +244,9 @@ func (l *Log) GetResource() map[string]any {
 
 // LogBatch represents a batch of logs
 type LogBatch struct {
-	Logs []*Log
-	Ack  func()
+	Logs       []*Log
+	Ack        func()
+	SampleType wal.SampleType
 }
 
 func (l *LogBatch) AddLiterals(literals []*LogLiteral) {
@@ -260,6 +262,7 @@ func (l *LogBatch) Reset() {
 	}
 	l.Logs = l.Logs[:0]
 	l.Ack = noop
+	l.SampleType = wal.UnknownSampleType
 }
 
 func noop() {}

collector/logs/types/logs_test.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/Azure/adx-mon/pkg/wal"
 	"github.com/stretchr/testify/require"
 )
 
@@ -149,6 +150,7 @@ func TestLogBatch(t *testing.T) {
 		Ack: func() {
 			ackCalled = true
 		},
+		SampleType: wal.HostLogSampleType,
 	}
 
 	require.Equal(t, 2, len(batch.Logs))
@@ -159,6 +161,7 @@ func TestLogBatch(t *testing.T) {
 	// Test Reset
 	batch.Reset()
 	require.Empty(t, batch.Logs)
+	require.Equal(t, wal.UnknownSampleType, batch.SampleType)
 }
 
 func BenchmarkLogWrites(b *testing.B) {

collector/otlp/logs_transfer.go

@@ -119,7 +119,7 @@ func (s *LogsService) Handler(w http.ResponseWriter, r *http.Request) {
 		if convertedLogs == 0 {
 			var status *status.Status
 			if droppedLogMissingMetadata > 0 {
-				metrics.InvalidLogsDropped.WithLabelValues().Add(float64(droppedLogMissingMetadata))
+				metrics.InvalidLogsDropped.WithLabelValues("missing_metadata").Add(float64(droppedLogMissingMetadata))
 				status = newErrorStatus("All logs dropped. Required kusto.database and kusto.table attributes or body fields are missing.")
 			} else {
 				status = newErrorStatus("No logs to process.")
@@ -138,7 +138,7 @@ func (s *LogsService) Handler(w http.ResponseWriter, r *http.Request) {
 				RejectedLogRecords: droppedLogMissingMetadata,
 				ErrorMessage:       "Logs lacking kube.database and kube.table attributes or body fields",
 			})
-			metrics.InvalidLogsDropped.WithLabelValues().Add(float64(droppedLogMissingMetadata))
+			metrics.InvalidLogsDropped.WithLabelValues("missing_metadata").Add(float64(droppedLogMissingMetadata))
 		}
 
 		respBodyBytes, err := proto.Marshal(resp)

docs/quick-start.md

@@ -91,6 +91,34 @@ AdxmonCollectorLogsSent
 | summarize TotalSent=sum(Value) by Host=tostring(Labels.host), Source=tostring(Labels.source)
 ```
 
+```kql
+// Estimate settled host-log loss by allowing uploads a grace period to arrive.
+let grace = 30m;
+let bucket = 1h;
+let start = ago(2d);
+let end = ago(grace);
+let collected =
+AdxmonCollectorLogsCommittedTotal
+| where Timestamp between (start .. end)
+| where Container == "collector"
+| invoke prom_delta()
+| extend Database=tostring(Labels.database), Table=tostring(Labels.table)
+| summarize Collected=sum(Value) by Bucket=bin(Timestamp, bucket), Database, Table;
+let uploaded =
+AdxmonIngestorLogsUploadedTotal
+| where Timestamp between (start + grace .. now())
+| where Container == "ingestor"
+| invoke prom_delta()
+| extend Database=tostring(Labels.database), Table=tostring(Labels.table)
+| summarize Uploaded=sum(Value) by Bucket=bin(Timestamp - grace, bucket), Database, Table;
+collected
+| join kind=fullouter uploaded on Bucket, Database, Table
+| extend Collected=coalesce(Collected, 0.0), Uploaded=coalesce(Uploaded, 0.0)
+| extend SettledLoss=iff(Collected > Uploaded, Collected - Uploaded, 0.0)
+| extend LossPct=iff(Collected > 0.0, 100.0 * SettledLoss / Collected, 0.0)
+| order by Bucket asc, Database asc, Table asc
+```
+
 ### Log Examples
 
 ```kql
@@ -132,4 +160,4 @@ Here's a glimpse of what comes as part of the pre-built dashboards:
 #### Namespaces
 ![Namespaces](images/namespaces-dashboard.png "Namespaces Dashboard")
 #### Pods
-![Pods](images/pods-dashboard.png "Pods Dashboard")
+![Pods](images/pods-dashboard.png "Pods Dashboard")

ingestor/adx/uploader.go

@@ -269,6 +269,25 @@ func sanitizeErrorString(err error) error {
 	return errors.New(errString)
 }
 
+func countSegmentSamplesByType(segmentReaders []*wal.SegmentReader, sampleType wal.SampleType) uint32 {
+	var total uint32
+	for _, sr := range segmentReaders {
+		st, sc := sr.SampleMetadata()
+		if st == sampleType {
+			total += sc
+		}
+	}
+	return total
+}
+
+func observeUploadedHostLogs(database, table string, segmentReaders []*wal.SegmentReader) {
+	hostLogCount := countSegmentSamplesByType(segmentReaders, wal.HostLogSampleType)
+	if hostLogCount == 0 {
+		return
+	}
+	metrics.IngestorLogsUploaded.WithLabelValues(database, table).Add(float64(hostLogCount))
+}
+
 func (n *uploader) upload(ctx context.Context) error {
 	for {
 		select {
@@ -383,6 +402,7 @@ func (n *uploader) upload(ctx context.Context) error {
 					logger.Errorf("Failed to remove batch: %s", err.Error())
 				}
 
+				observeUploadedHostLogs(database, table, segmentReaders)
 				metrics.IngestorSegmentsUploadedTotal.WithLabelValues(batch.Prefix).Add(float64(len(segmentReaders)))
 			}()
 

ingestor/adx/uploader_test.go

@@ -7,10 +7,13 @@ import (
 	"io"
 	"testing"
 
+	"github.com/Azure/adx-mon/metrics"
 	"github.com/Azure/adx-mon/pkg/testutils"
 	"github.com/Azure/adx-mon/pkg/testutils/kustainer"
+	"github.com/Azure/adx-mon/pkg/wal"
 	"github.com/Azure/azure-kusto-go/azkustodata"
 	kgzip "github.com/klauspost/compress/gzip"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stretchr/testify/require"
 	"github.com/testcontainers/testcontainers-go"
 )
@@ -72,3 +75,37 @@ func TestUploader_newGzipReader_PropagatesSourceError(t *testing.T) {
 	require.Error(t, err)
 	require.ErrorIs(t, err, expectedErr)
 }
+
+func TestObserveUploadedHostLogs(t *testing.T) {
+	metrics.IngestorLogsUploaded.Reset()
+
+	dir := t.TempDir()
+	hostReader := newSegmentReaderWithMetadata(t, dir, wal.HostLogSampleType, 2)
+	defer hostReader.Close()
+	otherReader := newSegmentReaderWithMetadata(t, dir, wal.LogSampleType, 5)
+	defer otherReader.Close()
+
+	observeUploadedHostLogs("Logs", "HostLogs", []*wal.SegmentReader{hostReader, otherReader})
+
+	var m dto.Metric
+	require.NoError(t, metrics.IngestorLogsUploaded.WithLabelValues("Logs", "HostLogs").Write(&m))
+	require.Equal(t, float64(2), m.GetCounter().GetValue())
+}
+
+func newSegmentReaderWithMetadata(t *testing.T, dir string, sampleType wal.SampleType, count uint32) *wal.SegmentReader {
+	t.Helper()
+
+	seg, err := wal.NewSegment(dir, "Logs_HostLogs")
+	require.NoError(t, err)
+
+	_, err = seg.Write(context.Background(), []byte("value\n"), wal.WithSampleMetadata(sampleType, count))
+	require.NoError(t, err)
+	path := seg.Path()
+	require.NoError(t, seg.Close())
+
+	reader, err := wal.NewSegmentReader(path)
+	require.NoError(t, err)
+	_, err = io.Copy(io.Discard, reader)
+	require.NoError(t, err)
+	return reader
+}

metrics/metrics.go

@@ -85,12 +85,19 @@ var (
 		Help:      "Counter of the number of segments uploaded to Kusto",
 	}, []string{"metric"})
 
+	IngestorLogsUploaded = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: Namespace,
+		Subsystem: "ingestor",
+		Name:      "logs_uploaded_total",
+		Help:      "Counter of host log records successfully uploaded to Kusto",
+	}, []string{"database", "table"})
+
 	InvalidLogsDropped = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: Namespace,
 		Subsystem: "ingestor",
 		Name:      "invalid_logs_dropped",
 		Help:      "Counter of the number of invalid logs dropped",
-	}, []string{})
+	}, []string{"reason"})
 
 	SampleLatency = promauto.NewGaugeVec(prometheus.GaugeOpts{
 		Namespace: Namespace,
@@ -185,6 +192,13 @@ var (
 		Help:      "Counter of the number of logs dropped due to errors",
 	}, []string{"source", "stage"})
 
+	CollectorLogsCommitted = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: Namespace,
+		Subsystem: "collector",
+		Name:      "logs_committed_total",
+		Help:      "Counter of host log records durably committed to the collector WAL",
+	}, []string{"database", "table"})
+
 	MetricsRequestsReceived = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: Namespace,
 		Subsystem: "collector",

pkg/wal/wal.go

@@ -113,6 +113,7 @@ const (
 	MetricSampleType
 	TraceSampleType
 	LogSampleType
+	HostLogSampleType
 )
 
 type WriteOptions func([]byte)
@@ -175,7 +176,7 @@ func (w *WAL) Write(ctx context.Context, buf []byte, opts ...WriteOptions) error
 	n, err := w.tryWrite(ctx, buf, opts...)
 	if errors.Is(err, ErrMaxSegmentSizeExceeded) {
 		w.rotateSegmentIfNecessary()
-		n, err = w.tryWrite(ctx, buf)
+		n, err = w.tryWrite(ctx, buf, opts...)
 		atomic.AddInt64(&w.segmentSize, int64(n))
 		return err
 	} else if errors.Is(err, ErrSegmentClosed) {

pkg/wal/wal_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"io"
 	"os"
 	"strings"
 	"testing"
@@ -182,6 +183,50 @@ func TestWAL_MaxSegmentCount(t *testing.T) {
 	}
 }
 
+func TestWAL_Write_PreservesSampleMetadataAfterRotation(t *testing.T) {
+	dir := t.TempDir()
+	const maxSegmentSize = 2048
+
+	w, err := wal.NewWAL(wal.WALOpts{
+		Prefix:         "Foo",
+		StorageDir:     dir,
+		SegmentMaxSize: maxSegmentSize,
+	})
+	require.NoError(t, err)
+	require.NoError(t, w.Open(context.Background()))
+
+	filler := make([]byte, 128)
+	_, err = rand.Read(filler)
+	require.NoError(t, err)
+
+	finalPayload := make([]byte, 512)
+	_, err = rand.Read(finalPayload)
+	require.NoError(t, err)
+
+	for w.Size()+len(filler) < maxSegmentSize && w.Size()+len(finalPayload) < maxSegmentSize {
+		require.NoError(t, w.Write(context.Background(), filler))
+	}
+
+	require.GreaterOrEqual(t, w.Size()+len(finalPayload), maxSegmentSize)
+
+	require.NoError(t, w.Write(context.Background(), finalPayload, wal.WithSampleMetadata(wal.HostLogSampleType, 1)))
+	require.NoError(t, w.Flush())
+
+	path := w.Path()
+	require.NoError(t, w.Close())
+
+	r, err := wal.NewSegmentReader(path)
+	require.NoError(t, err)
+	defer r.Close()
+
+	_, err = io.Copy(io.Discard, r)
+	require.NoError(t, err)
+
+	st, sc := r.SampleMetadata()
+	require.Equal(t, wal.HostLogSampleType, st)
+	require.Equal(t, uint32(1), sc)
+}
+
 func TestWAL_ActiveSegmentDiskUsageLeak(t *testing.T) {
 	dir := t.TempDir()
 	w, err := wal.NewWAL(wal.WALOpts{