Remove error logs in processing loop

Errors are provided as metrics only to minimize the impact on resources
consumption when an error is thrown repeatedly.

pkg/pipeline/ingest/ingest_grpc.go

@@ -105,7 +105,6 @@ func instrumentGRPC(m *metrics) grpc2.UnaryServerInterceptor {
 
 		resp, err = handler(ctx, req)
 		if err != nil {
-			glog.Errorf("Reporting metric error: %v", err)
 			m.error(fmt.Sprint(status.Code(err)))
 		}
 

pkg/pipeline/ingest/ingest_kafka.go

@@ -113,19 +113,17 @@ func (k *ingestKafka) isStopped() bool {
 }
 
 func (k *ingestKafka) processRecordDelay(record config.GenericMap) {
-	TimeFlowEndInterface, ok := record["TimeFlowEndMs"]
+	timeFlowEndInterface, ok := record["TimeFlowEndMs"]
 	if !ok {
-		klog.Errorf("TimeFlowEndMs missing in record %v", record)
 		k.metrics.error("TimeFlowEndMs missing")
 		return
 	}
-	TimeFlowEnd, ok := TimeFlowEndInterface.(int64)
+	timeFlowEnd, ok := timeFlowEndInterface.(int64)
 	if !ok {
-		klog.Errorf("Cannot parse TimeFlowEndMs of record %v", record)
 		k.metrics.error("Cannot parse TimeFlowEndMs")
 		return
 	}
-	delay := time.Since(time.UnixMilli(TimeFlowEnd)).Seconds()
+	delay := time.Since(time.UnixMilli(timeFlowEnd)).Seconds()
 	k.metrics.latency.Observe(delay)
 }