Add delta field counter

Author: Gregory Russell

metrics/metrics.go

@@ -34,6 +34,7 @@ func init() {
 	prometheus.MustRegister(BigQueryInsert)
 	prometheus.MustRegister(RowSizeHistogram)
 	prometheus.MustRegister(DeltaNumFieldsHistogram)
+	prometheus.MustRegister(SnapshotDeltaFieldCount)
 	prometheus.MustRegister(EntryFieldCountHistogram)
 	prometheus.MustRegister(DurationHistogram)
 	prometheus.MustRegister(InsertionHistogram)
@@ -277,6 +278,20 @@ var (
 		[]string{"table"},
 	)
 
+	// Count of fields occurring in snapshot deltas.
+	//
+	// Provides metrics:
+	//   etl_ndt_snapshot_delta_field_count_count
+	// Example usage:
+	//   metrics.SnapshotDeltaFieldCount.WithLabelValues(field.Name).Inc()
+	SnapshotDeltaFieldCount = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "etl_ndt_snapshot_delta_field_count_count",
+			Help: "Count of fields occurring in snapshot deltas.",
+		},
+		[]string{"field"},
+	)
+
 	// A histogram of snapshot delta field counts.  It is intended primarily for
 	// NDT.  Typical is about 13, but max might be up to 120 or so.
 	//

parser/ndt.go

@@ -348,6 +348,7 @@ func (n *NDTParser) processTest(test *fileInfoAndData, testType string) {
 }
 
 func (n *NDTParser) getDeltas(snaplog *web100.SnapLog, testType string) ([]schema.Web100ValueMap, int) {
+	// TODO - using make here causes insert error not detected by unit tests.
 	deltas := []schema.Web100ValueMap{}
 	deltaFieldCount := 0
 	if NDTOmitDeltas {

web100/web100.go

@@ -9,6 +9,8 @@ import (
 	"io"
 	"net"
 	"strings"
+
+	"github.com/m-lab/etl/metrics"
 )
 
 // NOTES:
@@ -553,7 +555,7 @@ func (snap *Snapshot) SnapshotValues(snapValues Saver) error {
 	return nil
 }
 
-// SnapshotValues writes changed values into the provided Saver.
+// SnapshotDeltas writes changed values into the provided Saver.
 func (snap *Snapshot) SnapshotDeltas(other *Snapshot, snapValues Saver) error {
 	if snap.raw == nil {
 		return errors.New("Empty/Invalid Snaplog")
@@ -568,6 +570,9 @@ func (snap *Snapshot) SnapshotDeltas(other *Snapshot, snapValues Saver) error {
 		a := other.raw[field.Offset : field.Offset+field.Size]
 		b := snap.raw[field.Offset : field.Offset+field.Size]
 		if bytes.Compare(a, b) != 0 {
+			if field.Name != "Duration" {
+				metrics.SnapshotDeltaFieldCount.WithLabelValues(field.Name).Inc()
+			}
 			field.Save(b, snapValues)
 		}
 	}