feat: add API for pipeline status check. Fixes #407. #599

dpadhiar · 2023-03-10T19:23:35Z

This PR adds an API for a pipeline status check.

The information returned currently is if the pipeline is "healthy" or not and a message accompanying this.

Pipeline health here is determined by checking the vertices of the pipeline and comparing their pending message counts to their processing rates. If the processing rate of a vertex of 0 but there are multiple pending messages left, this indicates there may be an issue with the vertex that should be checked.

Signed-off-by: Dillen Padhiar <[email protected]>

whynowy · 2023-03-10T23:26:41Z

pkg/apis/proto/daemon/daemon.proto

@@ -47,6 +47,12 @@ message VertexMetrics {
  map<string, int64> pendings = 4;
 }

+// PipelineStatus
+message PipelineStatus {
+  required bool health = 1;


We probably should go with a string type status, which has more meanings other than only true/false. For example, OK/Unknown/Error/...

whynowy · 2023-03-11T01:05:09Z

pkg/daemon/server/service/pipeline_metrics_query.go

+		vertexReq.Vertex = &vertex.Name
+		vertexResp, err := ps.GetVertexMetrics(ctx, vertexReq)
+		if err != nil {
+			return nil, err


If err is not nil, should not return an error, it probably just because autoscaling down to 0 and then metrics are not available. In this case, we could just return the status as Unknown.

whynowy · 2023-03-11T01:07:04Z

pkg/daemon/server/service/pipeline_metrics_query.go

+			return nil, err
+		}
+		// check default pending msg and processing rates
+		if vertexResp.VertexMetrics[0].Pendings["default"] > 0 && vertexResp.VertexMetrics[0].ProcessingRates["default"] == 0 {


Should not only check [0], reduce vertex has more items in the array.

Based on the GetVertexMetrics() implementation, default is not always available in the map, so also need to check if it's existing.

whynowy · 2023-03-11T01:07:37Z

pkg/daemon/server/service/pipeline_metrics_query.go

+		if vertexResp.VertexMetrics[0].Pendings["default"] > 0 && vertexResp.VertexMetrics[0].ProcessingRates["default"] == 0 {
+			resp.Status = &daemon.PipelineStatus{
+				Health:  pointer.Bool(false),
+				Message: pointer.String("Pipeline may have issue."),


More information in the message, such as which vertex has what issue.

Signed-off-by: Dillen Padhiar <[email protected]>

whynowy · 2023-03-14T21:49:38Z

pkg/apis/proto/daemon/daemon.proto

@@ -47,6 +47,12 @@ message VertexMetrics {
  map<string, int64> pendings = 4;
 }

+// PipelineStatus
+message PipelineStatus {
+  required string health = 1;


Use status?

whynowy · 2023-03-14T21:56:09Z

pkg/daemon/server/service/pipeline_metrics_query.go

+		for _, vertexMetrics := range vertexResp.VertexMetrics {
+			if pending, ok := vertexMetrics.Pendings["default"]; ok {
+				if processingRate, ok := vertexMetrics.ProcessingRates["default"]; ok {
+					if pending > 0 && processingRate == 0 {


Can you add some comments, that we need to revisit it later, it might also be caused by the processing is very slow, that the rate is 0.

whynowy · 2023-03-14T21:57:00Z

pkg/daemon/server/service/pipeline_metrics_query.go

+		if err != nil {
+			resp.Status = &daemon.PipelineStatus{
+				Health:  pointer.String(metrics.PipelineStatusUnknown),
+				Message: pointer.String("Pipeline health is unknown."),


"Pipeline status is unknown."

whynowy · 2023-03-14T22:02:29Z

pkg/metrics/metrics.go

@@ -46,6 +46,10 @@ const (

 	VertexProcessingRate  = "vertex_processing_rate"
 	VertexPendingMessages = "vertex_pending_messages"
+
+	PipelineStatusOK      = "OK"


Can we move them to pipeline_metrics_query.go for now?

Signed-off-by: Dillen Padhiar <[email protected]>

whynowy · 2023-03-14T22:56:37Z

pkg/apis/proto/daemon/daemon.proto

@@ -110,4 +124,9 @@ service DaemonService {
  rpc GetPipelineWatermarks (GetPipelineWatermarksRequest) returns (GetPipelineWatermarksResponse) {
    option (google.api.http).get = "/api/v1/pipelines/{pipeline}/watermarks";
  };
+
+  // GetPipelineStatus () returns (PipelineStatus) // status: bool and string


Comment is not up to date.

whynowy · 2023-03-14T22:57:32Z

pkg/daemon/server/service/pipeline_metrics_query.go

+					}
+				}
+			}
+			if vertexMetrics.Pendings["1m"] > 0 && vertexMetrics.ProcessingRates["1m"] == 0 {


What I meant was, we don't need to check 1m.

Got it, for future reference, what are the cases "default" won't be in the map and why?

The default will be always in the map if 1m is there. But if you look at the GetVertexMetrics implementation, there's chance the map is empty when the target pod is scaled down to 0 or something is wrong with that pod.

Signed-off-by: Dillen Padhiar <[email protected]>

whynowy · 2023-03-15T04:51:03Z

pkg/daemon/server/service/pipeline_metrics_query.go

+						return resp, nil
+					}
+				}
+			} else {


We don't need this else.

It is expected sometimes there's no pending information:

Source like HTTP, it does not have pending at all;

Based on the GetVertexMetrics implementation, when scaling down to 0 replica, the pending is also not available.

For these cases, we should consider it's ok.

whynowy · 2023-03-15T04:52:40Z

@dpadhiar - please also test this api with real pipelines to make sure the output is expected (find the pipeline spec in examples).

dpadhiar · 2023-03-15T05:02:53Z

@dpadhiar - please also test this api with real pipelines to make sure the output is expected (find the pipeline spec in examples).

Will do, I will test the API with most of them tomorrow morning.

Signed-off-by: Dillen Padhiar <[email protected]>

… issue-407

Signed-off-by: Dillen Padhiar <[email protected]>

dpadhiar and others added 4 commits March 9, 2023 16:27

feat: add API for pipeline status check

62f5295

Signed-off-by: Dillen Padhiar <[email protected]>

test: add test case for pipeline metrics query

dbe23a5

Signed-off-by: Dillen Padhiar <[email protected]>

chore: remove comments from proto

558a346

Signed-off-by: Dillen Padhiar <[email protected]>

Merge branch 'main' into issue-407

0cb57b0

whynowy reviewed Mar 11, 2023

View reviewed changes

dpadhiar and others added 2 commits March 13, 2023 14:13

feat: update error message to say which vertex has issue

b8aa572

Signed-off-by: Dillen Padhiar <[email protected]>

Merge branch 'main' into issue-407

417db28

dpadhiar requested a review from whynowy March 13, 2023 21:13

dpadhiar marked this pull request as ready for review March 13, 2023 21:52

dpadhiar requested a review from vigith as a code owner March 13, 2023 21:52

Merge branch 'main' into issue-407

a6e0307

whynowy reviewed Mar 14, 2023

View reviewed changes

feat: refactor and add comments

e304de2

Signed-off-by: Dillen Padhiar <[email protected]>

dpadhiar requested review from whynowy and removed request for vigith March 14, 2023 22:36

whynowy reviewed Mar 14, 2023

View reviewed changes

dpadhiar added 2 commits March 14, 2023 17:07

feat: refactor and update

7dfd2fa

Signed-off-by: Dillen Padhiar <[email protected]>

test: fix test case

f6773c2

Signed-off-by: Dillen Padhiar <[email protected]>

whynowy reviewed Mar 15, 2023

View reviewed changes

dpadhiar and others added 3 commits March 15, 2023 08:27

feat: remove else statement for unnecessary check

ff5c66f

Signed-off-by: Dillen Padhiar <[email protected]>

Merge branch 'main' into issue-407

2b0e07a

Merge branch 'issue-407' of https://github.com/dpadhiar/numaflow into…

db04a38

… issue-407

dpadhiar requested a review from whynowy March 15, 2023 15:53

whynowy approved these changes Mar 15, 2023

View reviewed changes

whynowy merged commit c981513 into numaproj:main Mar 15, 2023

whynowy pushed a commit that referenced this pull request Apr 3, 2023

feat: add API for pipeline status check. Fixes #407. (#599)

a155f2a

Signed-off-by: Dillen Padhiar <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add API for pipeline status check. Fixes #407. #599

feat: add API for pipeline status check. Fixes #407. #599

dpadhiar commented Mar 10, 2023

whynowy Mar 10, 2023

whynowy Mar 11, 2023

whynowy Mar 11, 2023

whynowy Mar 11, 2023

whynowy Mar 14, 2023

whynowy Mar 14, 2023

whynowy Mar 14, 2023

whynowy Mar 14, 2023

whynowy Mar 14, 2023

whynowy Mar 14, 2023

dpadhiar Mar 14, 2023

whynowy Mar 14, 2023

whynowy Mar 15, 2023

whynowy commented Mar 15, 2023

dpadhiar commented Mar 15, 2023

feat: add API for pipeline status check. Fixes #407. #599

feat: add API for pipeline status check. Fixes #407. #599

Conversation

dpadhiar commented Mar 10, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

whynowy commented Mar 15, 2023

dpadhiar commented Mar 15, 2023