Ensure that metric id is always set during kafka ingestion

cmd/mt-gateway/ingest/metrics.go

@@ -89,6 +89,7 @@ func prepareIngest(in []*schema.MetricData, toPublish []*schema.MetricData) ([]*
 			promDiscards.Add(m.OrgId, err.Error())
 			continue
 		}
+		m.SetId()
 		metricTimestamp = getMetricsTimestampStat(m.OrgId)
 		metricTimestamp.ValueUint32(uint32(m.Time))
 		toPublish = append(toPublish, m)

publish/kafka/publish.go

@@ -276,14 +276,12 @@ func (m *mtPublisher) Publish(metrics []*schema.MetricData) error {
 
 	for _, metric := range metrics {
 		if metric.Interval == 0 {
-			if m.autoInterval {
-				_, s := m.schemas.Match(metric.Name, 0)
-				metric.Interval = s.Retentions.Rets[0].SecondsPerPoint
-				metric.SetId()
-			} else {
-				log.Error("interval is 0 but can't deduce interval automatically. this should never happen")
-				return errors.New("need to deduce interval but cannot")
-			}
+			log.Error("interval must be set before publishing")
+			return errors.New("interval must be set before publishing")
+		}
+		if metric.Id == "" {
+			log.Error("metric ID must be set before publishing")
+			return errors.New("metric ID must be set before publishing")
 		}
 
 		isMD := false

publish/publish.go

@@ -18,7 +18,10 @@ var (
 )
 
 type Publisher interface {
+	//Publish the given metrics.
+	//Requires that the metric interval and ID have been set
 	Publish(metrics []*schema.MetricData) error
+	//Type returns the type of system the Publisher publishes to
 	Type() string
 }