<commit_before>package main

import (
	"fmt"
	"math"
	"net/http"
	"strconv"
	"sync/atomic"
	"time"

	"github.com/grafana/metrictank/clock"
	"github.com/grafana/metrictank/stacktest/graphite"
	"github.com/grafana/metrictank/stats"
	"github.com/grafana/metrictank/util/align"
	log "github.com/sirupsen/logrus"
)

var (
	httpError    = stats.NewCounter32WithTags("parrot.monitoring.error", ";error=http")
	decodeError  = stats.NewCounter32WithTags("parrot.monitoring.error", ";error=decode")
	invalidError = stats.NewCounter32WithTags("parrot.monitoring.error", ";error=invalid")
)

var metricsBySeries []partitionMetrics

func initMetricsBySeries() {
	for p := 0; p < int(partitionCount); p++ {
		metricsBySeries = append(metricsBySeries, NewPartitionMetrics(p))
	}
}

type seriesInfo struct {
	lastTs uint32 // last timestamp seen in the response

	// to generate stats from
	lastSeen         uint32  // the last seen non-NaN time stamp (to generate lag from)
	deltaSum         float64 // sum of abs(value - ts) across the time series
	numNans          int32   // number of missing values for each series
	numNonMatching   int32   // number of points where value != ts
	correctNumPoints bool    // whether the expected number of points were received
	correctAlignment bool    // whether the last ts matches `now`
	correctSpacing   bool    // whether all points are sorted and 1 period apart
}

type partitionMetrics struct {
	lag              *stats.Gauge32 // time since the last value was recorded
	deltaSum         *stats.Gauge32 // total amount of drift between expected value and actual values
	numNans          *stats.Gauge32 // number of missing values for each series
	numNonMatching   *stats.Gauge32 // number of points where value != ts
	correctNumPoints *stats.Bool    // whether the expected number of points were received
	correctAlignment *stats.Bool    // whether the last ts matches `now`
	correctSpacing   *stats.Bool    // whether all points are sorted and 1 period apart
}

func NewPartitionMetrics(p int) partitionMetrics {
	return partitionMetrics{
		//TODO enable metrics2docs by adding 'metric' prefix to each metric
		// parrot.monitoring.lag is the time since the last value was recorded
		lag: stats.NewGauge32WithTags("parrot.monitoring.lag", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.deltaSum is the total amount of drift between expected value and actual values
		deltaSum: stats.NewGauge32WithTags("parrot.monitoring.deltaSum", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.nans is the number of missing values for each series
		numNans: stats.NewGauge32WithTags("parrot.monitoring.nans", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.nonmatching is the total number of entries where drift occurred
		numNonMatching: stats.NewGauge32WithTags("parrot.monitoring.nonmatching", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.correctNumPoints is whether the expected number of points were received
		correctNumPoints: stats.NewBoolWithTags("parrot.monitoring.correctNumPoints", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.correctAlignment is whether the last ts matches `now`
		correctAlignment: stats.NewBoolWithTags("parrot.monitoring.correctAlignment", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.correctSpacing is whether all points are sorted and 1 period apart
		correctSpacing: stats.NewBoolWithTags("parrot.monitoring.correctSpacing", fmt.Sprintf(";partition=%d", p)),
	}
}

func monitor() {
	initMetricsBySeries()
	for tick := range clock.AlignedTickLossy(queryInterval) {

		resp := graphite.ExecuteRenderQuery(buildRequest(tick))
		if resp.HTTPErr != nil {
			httpError.Inc()
			continue
		}
		if resp.DecodeErr != nil {
			decodeError.Inc()
			continue
		}

		var invalid bool
		for _, s := range resp.Decoded {
			ok := processPartitionSeries(s, tick)
			if !ok {
				invalid = true
			}
		}
		if invalid {
			invalidError.Inc()
		}
		statsGraphite.Report(tick)
	}
}

func processPartitionSeries(s graphite.Series, now time.Time) bool {
	partition, err := strconv.Atoi(s.Target)
	if err != nil {
		log.Debug("unable to parse partition", err)
		return false
	}

	serStats := seriesInfo{}
	lastTs := align.Forward(uint32(now.Unix()), uint32(testMetricsInterval.Seconds()))
	serStats.lastTs = s.Datapoints[len(s.Datapoints)-1].Ts
	serStats.correctAlignment = serStats.lastTs == lastTs
	serStats.correctNumPoints = len(s.Datapoints) == int(lookbackPeriod/testMetricsInterval)
	serStats.correctSpacing = checkSpacing(s.Datapoints)

	for _, dp := range s.Datapoints {
		if math.IsNaN(dp.Val) {
			serStats.numNans += 1
			continue
		}
		serStats.lastSeen = dp.Ts
		if diff := dp.Val - float64(dp.Ts); diff != 0 {
			log.Debugf("partition=%d dp.Val=%f dp.Ts=%d diff=%f", partition, dp.Val, dp.Ts, diff)
			serStats.deltaSum += diff
			serStats.numNonMatching += 1
		}
	}

	metrics := metricsBySeries[partition]
	metrics.numNans.Set(int(serStats.numNans))
	lag := atomic.LoadInt64(&lastPublish) - int64(serStats.lastSeen)
	metrics.lag.Set(int(lag))
	metrics.deltaSum.Set(int(serStats.deltaSum))
	metrics.numNonMatching.Set(int(serStats.numNonMatching))
	metrics.correctNumPoints.Set(serStats.correctNumPoints)
	metrics.correctAlignment.Set(serStats.correctAlignment)
	metrics.correctSpacing.Set(serStats.correctSpacing)

	return true
}
<commit_msg>don't round deltaSum down. we need to know every difference<commit_after>package main

import (
	"fmt"
	"math"
	"net/http"
	"strconv"
	"sync/atomic"
	"time"

	"github.com/grafana/metrictank/clock"
	"github.com/grafana/metrictank/stacktest/graphite"
	"github.com/grafana/metrictank/stats"
	"github.com/grafana/metrictank/util/align"
	log "github.com/sirupsen/logrus"
)

var (
	httpError    = stats.NewCounter32WithTags("parrot.monitoring.error", ";error=http")
	decodeError  = stats.NewCounter32WithTags("parrot.monitoring.error", ";error=decode")
	invalidError = stats.NewCounter32WithTags("parrot.monitoring.error", ";error=invalid")
)

var metricsBySeries []partitionMetrics

func initMetricsBySeries() {
	for p := 0; p < int(partitionCount); p++ {
		metricsBySeries = append(metricsBySeries, NewPartitionMetrics(p))
	}
}

type seriesInfo struct {
	lastTs uint32 // last timestamp seen in the response

	// to generate stats from
	lastSeen         uint32  // the last seen non-NaN time stamp (to generate lag from)
	deltaSum         float64 // sum of abs(value - ts) across the time series
	numNans          int32   // number of missing values for each series
	numNonMatching   int32   // number of points where value != ts
	correctNumPoints bool    // whether the expected number of points were received
	correctAlignment bool    // whether the last ts matches `now`
	correctSpacing   bool    // whether all points are sorted and 1 period apart
}

type partitionMetrics struct {
	lag              *stats.Gauge32 // time since the last value was recorded
	deltaSum         *stats.Gauge32 // total amount of drift between expected value and actual values
	numNans          *stats.Gauge32 // number of missing values for each series
	numNonMatching   *stats.Gauge32 // number of points where value != ts
	correctNumPoints *stats.Bool    // whether the expected number of points were received
	correctAlignment *stats.Bool    // whether the last ts matches `now`
	correctSpacing   *stats.Bool    // whether all points are sorted and 1 period apart
}

func NewPartitionMetrics(p int) partitionMetrics {
	return partitionMetrics{
		//TODO enable metrics2docs by adding 'metric' prefix to each metric
		// parrot.monitoring.lag is the time since the last value was recorded
		lag: stats.NewGauge32WithTags("parrot.monitoring.lag", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.deltaSum is the total amount of drift between expected value and actual values
		deltaSum: stats.NewGauge32WithTags("parrot.monitoring.deltaSum", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.nans is the number of missing values for each series
		numNans: stats.NewGauge32WithTags("parrot.monitoring.nans", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.nonmatching is the total number of entries where drift occurred
		numNonMatching: stats.NewGauge32WithTags("parrot.monitoring.nonmatching", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.correctNumPoints is whether the expected number of points were received
		correctNumPoints: stats.NewBoolWithTags("parrot.monitoring.correctNumPoints", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.correctAlignment is whether the last ts matches `now`
		correctAlignment: stats.NewBoolWithTags("parrot.monitoring.correctAlignment", fmt.Sprintf(";partition=%d", p)),
		// parrot.monitoring.correctSpacing is whether all points are sorted and 1 period apart
		correctSpacing: stats.NewBoolWithTags("parrot.monitoring.correctSpacing", fmt.Sprintf(";partition=%d", p)),
	}
}

func monitor() {
	initMetricsBySeries()
	for tick := range clock.AlignedTickLossy(queryInterval) {

		resp := graphite.ExecuteRenderQuery(buildRequest(tick))
		if resp.HTTPErr != nil {
			httpError.Inc()
			continue
		}
		if resp.DecodeErr != nil {
			decodeError.Inc()
			continue
		}

		var invalid bool
		for _, s := range resp.Decoded {
			ok := processPartitionSeries(s, tick)
			if !ok {
				invalid = true
			}
		}
		if invalid {
			invalidError.Inc()
		}
		statsGraphite.Report(tick)
	}
}

func processPartitionSeries(s graphite.Series, now time.Time) bool {
	partition, err := strconv.Atoi(s.Target)
	if err != nil {
		log.Debug("unable to parse partition", err)
		return false
	}

	serStats := seriesInfo{}
	lastTs := align.Forward(uint32(now.Unix()), uint32(testMetricsInterval.Seconds()))
	serStats.lastTs = s.Datapoints[len(s.Datapoints)-1].Ts
	serStats.correctAlignment = serStats.lastTs == lastTs
	serStats.correctNumPoints = len(s.Datapoints) == int(lookbackPeriod/testMetricsInterval)
	serStats.correctSpacing = checkSpacing(s.Datapoints)

	for _, dp := range s.Datapoints {
		if math.IsNaN(dp.Val) {
			serStats.numNans += 1
			continue
		}
		serStats.lastSeen = dp.Ts
		if diff := dp.Val - float64(dp.Ts); diff != 0 {
			log.Debugf("partition=%d dp.Val=%f dp.Ts=%d diff=%f", partition, dp.Val, dp.Ts, diff)
			serStats.deltaSum += diff
			serStats.numNonMatching += 1
		}
	}

	metrics := metricsBySeries[partition]
	metrics.numNans.Set(int(serStats.numNans))
	lag := atomic.LoadInt64(&lastPublish) - int64(serStats.lastSeen)
	metrics.lag.Set(int(lag))
	metrics.deltaSum.Set(int(math.Ceil(serStats.deltaSum)))
	metrics.numNonMatching.Set(int(serStats.numNonMatching))
	metrics.correctNumPoints.Set(serStats.correctNumPoints)
	metrics.correctAlignment.Set(serStats.correctAlignment)
	metrics.correctSpacing.Set(serStats.correctSpacing)

	return true
}

{'fraction_non_alphanumeric': 0.11144158379031509, 'fraction_numerical': 0.009480434984664003, 'mean_word_length': 4.3585657370517925, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1517221', 'n_tokens_mistral': 3592, 'n_tokens_neox': 3359, 'n_words': 853}