show error on invalid metric name (#430)

* show error on invalid metric name

* fmt

* fix tests

* add test

Author: dinal

pkg/appender/appender.go

@@ -131,7 +131,7 @@ type MetricsCache struct {
 	lastError           error
 	performanceReporter *performance.MetricReporter
 
-	stopChan      chan int
+	stopChan chan int
 }
 
 func NewMetricsCache(container v3io.Container, logger logger.Logger, cfg *config.V3ioConfig,
@@ -219,6 +219,10 @@ func (mc *MetricsCache) Add(lset utils.LabelsIfc, t int64, v interface{}) (uint6
 	}
 
 	name, key, hash := lset.GetKey()
+	err = utils.IsValidMetricName(name)
+	if err != nil {
+		return 0, err
+	}
 	metric, ok := mc.getMetric(name, hash)
 
 	var aggrMetrics []*MetricState

pkg/pquerier/collector.go

@@ -66,7 +66,11 @@ func mainCollector(ctx *selectQueryContext, responseChannel chan *qryResults) {
 
 	for res := range responseChannel {
 		if res.IsRawQuery() {
-			rawCollector(ctx, res)
+			err := rawCollector(ctx, res)
+			if err != nil {
+				ctx.errorChannel <- err
+				return
+			}
 		} else {
 			err := res.frame.addMetricIfNotExist(res.name, ctx.getResultBucketsSize(), res.IsServerAggregates())
 			if err != nil {
@@ -104,7 +108,7 @@ func mainCollector(ctx *selectQueryContext, responseChannel chan *qryResults) {
 	}
 }
 
-func rawCollector(ctx *selectQueryContext, res *qryResults) {
+func rawCollector(ctx *selectQueryContext, res *qryResults) error {
 	ctx.logger.Debug("using Raw Collector for metric %v", res.name)
 
 	if res.frame.isWildcardSelect {
@@ -114,8 +118,7 @@ func rawCollector(ctx *selectQueryContext, res *qryResults) {
 		} else {
 			series, err := NewRawSeries(res, ctx.logger.GetChild("v3ioRawSeries"))
 			if err != nil {
-				ctx.errorChannel <- err
-				return
+				return err
 			}
 			res.frame.rawColumns = append(res.frame.rawColumns, series)
 			res.frame.columnByName[res.name] = len(res.frame.rawColumns) - 1
@@ -128,12 +131,12 @@ func rawCollector(ctx *selectQueryContext, res *qryResults) {
 		} else {
 			series, err := NewRawSeries(res, ctx.logger.GetChild("v3ioRawSeries"))
 			if err != nil {
-				ctx.errorChannel <- err
-				return
+				return err
 			}
 			res.frame.rawColumns[columnIndex] = series
 		}
 	}
+	return nil
 }
 
 func aggregateClientAggregates(ctx *selectQueryContext, res *qryResults) {

pkg/pquerier/pqueriertest/raw_query_integration_test.go

@@ -233,42 +233,6 @@ func (suite *testRawQuerySuite) TestQueryWithBadTimeParameters() {
 	}
 }
 
-func (suite *testRawQuerySuite) TestQueryMetricWithDashInTheName() { // IG-8585
-	adapter, err := tsdb.NewV3ioAdapter(suite.v3ioConfig, nil, nil)
-	if err != nil {
-		suite.T().Fatalf("failed to create v3io adapter. reason: %s", err)
-	}
-
-	labels1 := utils.LabelsFromStringList("os", "linux")
-	numberOfEvents := 10
-	eventsInterval := 60 * 1000
-
-	expectedData := []tsdbtest.DataPoint{{suite.basicQueryTime, 10},
-		{int64(suite.basicQueryTime + tsdbtest.MinuteInMillis), 20},
-		{suite.basicQueryTime + 2*tsdbtest.MinuteInMillis, 30},
-		{suite.basicQueryTime + 3*tsdbtest.MinuteInMillis, 40}}
-	testParams := tsdbtest.NewTestParams(suite.T(),
-		tsdbtest.TestOption{
-			Key: tsdbtest.OptTimeSeries,
-			Value: tsdbtest.TimeSeries{tsdbtest.Metric{
-				Name:   "cool-cpu",
-				Labels: labels1,
-				Data:   expectedData},
-			}})
-	tsdbtest.InsertData(suite.T(), testParams)
-
-	querierV2, err := adapter.QuerierV2()
-	if err != nil {
-		suite.T().Fatalf("Failed to create querier v2, err: %v", err)
-	}
-
-	params := &pquerier.SelectParams{From: suite.basicQueryTime, To: suite.basicQueryTime + int64(numberOfEvents*eventsInterval)}
-	_, err = querierV2.Select(params)
-	if err == nil {
-		suite.T().Fatalf("expected an error but finish succesfully")
-	}
-}
-
 func (suite *testRawQuerySuite) TestSelectRawDataByRequestedColumns() {
 	adapter, err := tsdb.NewV3ioAdapter(suite.v3ioConfig, nil, nil)
 	if err != nil {

pkg/tsdb/v3iotsdb_integration_test.go

@@ -89,26 +89,12 @@ func TestIngestData(t *testing.T) {
 					}}},
 			),
 		},
-		{desc: "Should ingest record with a dash in the metric name (IG-8585)",
-			params: tsdbtest.NewTestParams(t,
-				tsdbtest.TestOption{
-					Key: tsdbtest.OptTimeSeries,
-					Value: tsdbtest.TimeSeries{tsdbtest.Metric{
-						Name:   "cool-cpu",
-						Labels: utils.LabelsFromStringList("os", "linux", "iguaz", "yesplease"),
-						Data: []tsdbtest.DataPoint{
-							{Time: 1532940510, Value: 314.3},
-							{Time: 1532940510 + 5, Value: 300.3},
-							{Time: 1532940510 - 10, Value: 3234.6}},
-					}}},
-			),
-		},
 		{desc: "Should ingest into first partition in epoch without corruption (TSDB-67)",
 			params: tsdbtest.NewTestParams(t,
 				tsdbtest.TestOption{
 					Key: tsdbtest.OptTimeSeries,
 					Value: tsdbtest.TimeSeries{tsdbtest.Metric{
-						Name:   "cool-cpu",
+						Name:   "coolcpu",
 						Labels: utils.LabelsFromStringList("os", "linux", "iguaz", "yesplease"),
 						Data: []tsdbtest.DataPoint{
 							{Time: 10, Value: 314.3},
@@ -216,6 +202,37 @@ func TestIngestDataWithSameTimestamp(t *testing.T) {
 	tsdbtest.ValidateCountOfSamples(t, adapter, "", 2, baseTime-1*tsdbtest.HoursInMillis, baseTime+1*tsdbtest.HoursInMillis, -1)
 }
 
+func TestWriteMetricWithDashInName(t *testing.T) {
+	testParams := tsdbtest.NewTestParams(t,
+		tsdbtest.TestOption{
+			Key: tsdbtest.OptTimeSeries,
+			Value: tsdbtest.TimeSeries{tsdbtest.Metric{
+				Name:   "cpu-1",
+				Labels: utils.LabelsFromStringList("testLabel", "balbala"),
+				Data:   []tsdbtest.DataPoint{{Time: 1532940510, Value: 314.3}},
+			}}})
+	defer tsdbtest.SetUp(t, testParams)()
+
+	adapter, err := NewV3ioAdapter(testParams.V3ioConfig(), nil, nil)
+	if err != nil {
+		t.Fatalf("Failed to create v3io adapter. reason: %s", err)
+	}
+
+	appender, err := adapter.Appender()
+	if err != nil {
+		t.Fatalf("Failed to get appender. reason: %s", err)
+	}
+	for _, dp := range testParams.TimeSeries() {
+		labels := utils.Labels{utils.Label{Name: "__name__", Value: dp.Name}}
+		labels = append(labels, dp.Labels...)
+
+		_, err := appender.Add(labels, dp.Data[0].Time, dp.Data[0].Value)
+		if err == nil {
+			t.Fatalf("Test should have failed")
+		}
+	}
+}
+
 func TestQueryData(t *testing.T) {
 	testCases := []struct {
 		desc         string
@@ -296,21 +313,6 @@ func TestQueryData(t *testing.T) {
 			step:     defaultStepMs,
 			expected: map[string][]tsdbtest.DataPoint{"": {{Time: 1532940510, Value: 31.3}}}},
 
-		{desc: "Should ingest and query data with '-' in the metric name (IG-8585)",
-			testParams: tsdbtest.NewTestParams(t,
-				tsdbtest.TestOption{
-					Key: tsdbtest.OptTimeSeries,
-					Value: tsdbtest.TimeSeries{tsdbtest.Metric{
-						Name:   "cool-cpu",
-						Labels: utils.LabelsFromStringList("os", "linux", "iguaz", "yesplease"),
-						Data:   []tsdbtest.DataPoint{{Time: 1532940510, Value: 314.3}},
-					}}},
-			),
-			from:     0,
-			to:       1532940510 + 1,
-			step:     defaultStepMs,
-			expected: map[string][]tsdbtest.DataPoint{"": {{Time: 1532940510, Value: 314.3}}}},
-
 		{desc: "Should ingest and query by time",
 			testParams: tsdbtest.NewTestParams(t,
 				tsdbtest.TestOption{