fixes3

Signed-off-by: Samhith Kakarla <[email protected]>

docs/operations/metrics/metrics.md

@@ -59,6 +59,7 @@ These metrics can be used to determine the latency of your pipeline.
 
 | Metric name                                    | Metric type | Labels                                                                                                                                                        | Description                                                                                    |
 | ---------------------------------------------- | ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- |------------------------------------------------------------------------------------------------|
+| `pipeline_lag_milliseconds` | Gauge| `pipeline=<pipeline-name>` | Provides the pipeline processing lag in milliseconds |
 | `source_forwarder_transformer_processing_time` | Histogram   | `pipeline=<pipeline-name>` <br> `vertex=<vertex-name>` <br> `vertex_type=<vertex-type>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of User-defined Source Transformer   |
 | `forwarder_udf_processing_time`                | Histogram   | `pipeline=<pipeline-name>` <br> `vertex=<vertex-name>` <br> `vertex_type=<vertex-type>` <br> `replica=<replica-index>`                                        | Provides a histogram distribution of the processing times of User-defined Functions. (UDF's)   |
 | `forwarder_forward_chunk_processing_time`      | Histogram   | `pipeline=<pipeline-name>` <br> `vertex=<vertex-name>` <br> `vertex_type=<vertex-type>` <br> `replica=<replica-index>`                                        | Provides a histogram distribution of the processing times of the forwarder function as a whole |

pkg/daemon/server/metrics.go

@@ -93,9 +93,13 @@ func (ds *daemonServer) exposeLagMetrics(ctx context.Context) {
 			// if the data hasn't arrived the sink vertex
 			// set the lag to be -1
 			if minWM < 0 {
-				pipelineProcessingLag.WithLabelValues(metrics.LabelPipeline).Set(-1)
+				pipelineProcessingLag.WithLabelValues(ds.pipeline.Name).Set(-1)
 			} else {
-				pipelineProcessingLag.WithLabelValues(metrics.LabelPipeline).Set(float64(maxWM - minWM))
+				if maxWM < minWM {
+					pipelineProcessingLag.WithLabelValues(ds.pipeline.Name).Set(-1)
+				} else {
+					pipelineProcessingLag.WithLabelValues(ds.pipeline.Name).Set(float64(maxWM - minWM))
+				}
 			}
 		case <-ctx.Done():
 			return