Repository: numaproj/numaflow
Pull request #1406 (2023-12-06 10:11:31)

feat: health status implementation
Health status definitions 

It is divided into two parts:
1. Pipeline Resource Health: It is based on the health of each vertex in the pipeline
2. Data Criticality: It is based on the data movement of the pipeline

**Resource Health** can be "healthy (0) | unhealthy (1) | paused (3) | unknown (4)".

Resource Health purely means it is up and running.
Resource health will be the max(health) based of each vertex's health

Resource health checks if all the pods are in running state for the pipeline, and also for paused, unknown pipeline etc

**Data Criticality** on the other end shows whether the pipeline is working as expected.
It represents the pending messages, lags, etc.
Data Criticality can be "ok (0) | warning (1) | critical (2)".
A backlogged pipeline can be healthy even though it has an increasing back-pressure.

For data criticality the timeline data is populated for the pipeline, and if the average usage lies above the thresholds then the required state is assigned. For critical states we have the option do a lookback and only assign it to be critical if we see a predefined number of critical state in a lookback period window. 






whynowy: @kohlisid - resolve the conflicts?

Patch:
diff --git a/go.mod b/go.mod
index 3232884e..47479a04 100644
--- a/go.mod
+++ b/go.mod
@@ -45,6 +45,7 @@ require (
 	go.uber.org/multierr v1.7.0
 	go.uber.org/zap v1.24.0
 	golang.org/x/crypto v0.17.0
+	golang.org/x/net v0.17.0
 	golang.org/x/oauth2 v0.7.0
 	golang.org/x/sync v0.3.0
 	google.golang.org/genproto/googleapis/api v0.0.0-20230525234035-dd9d682886f9
@@ -190,7 +191,6 @@ require (
 	go.mongodb.org/mongo-driver v1.7.3 // indirect
 	golang.org/x/arch v0.3.0 // indirect
 	golang.org/x/mod v0.8.0 // indirect
-	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/sys v0.15.0 // indirect
 	golang.org/x/term v0.15.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
diff --git a/pkg/apis/proto/daemon/daemon.pb.go b/pkg/apis/proto/daemon/daemon.pb.go
index 58d8d8e2..107cdfa9 100644
--- a/pkg/apis/proto/daemon/daemon.pb.go
+++ b/pkg/apis/proto/daemon/daemon.pb.go
@@ -220,6 +220,7 @@ func (m *VertexMetrics) GetPendings() map[string]int64 {
 type PipelineStatus struct {
 	Status               *string  `protobuf:"bytes,1,req,name=status" json:"status,omitempty"`
 	Message              *string  `protobuf:"bytes,2,req,name=message" json:"message,omitempty"`
+	Code                 *string  `protobuf:"bytes,3,req,name=code" json:"code,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -272,6 +273,13 @@ func (m *PipelineStatus) GetMessage() string {
 	return ""
 }
 
+func (m *PipelineStatus) GetCode() string {
+	if m != nil && m.Code != nil {
+		return *m.Code
+	}
+	return ""
+}
+
 type ListBuffersRequest struct {
 	Pipeline             *string  `protobuf:"bytes,1,req,name=pipeline" json:"pipeline,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -871,65 +879,66 @@ func init() {
 }
 
 var fileDescriptor_93e327fd0d673221 = []byte{
-	// 926 bytes of a gzipped FileDescriptorProto
+	// 937 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x56, 0x5f, 0x6f, 0xdc, 0x44,
 	0x10, 0x97, 0x7d, 0xf9, 0x73, 0x37, 0xe1, 0xda, 0x30, 0xb4, 0xc1, 0x75, 0x4b, 0x30, 0x6e, 0x0a,
-	0x47, 0x28, 0x67, 0x88, 0x44, 0x55, 0xb5, 0x12, 0x45, 0x29, 0x69, 0x84, 0x48, 0x50, 0xb4, 0x05,
-	0x2a, 0xf1, 0xe6, 0xdc, 0xed, 0xb9, 0x26, 0xb6, 0xd7, 0x78, 0xd7, 0x17, 0xa2, 0x2a, 0x2f, 0x48,
-	0x7c, 0x02, 0xd4, 0x4f, 0x82, 0xc4, 0x47, 0x40, 0x3c, 0x22, 0xf1, 0xc8, 0x0b, 0x8a, 0xf8, 0x20,
-	0x95, 0x77, 0xd7, 0x57, 0xfb, 0xe2, 0xbb, 0xe4, 0x29, 0x3b, 0x33, 0xbf, 0x99, 0xf9, 0x79, 0xfe,
-	0x5d, 0xc0, 0x4d, 0x8f, 0x02, 0xcf, 0x4f, 0x43, 0xee, 0xa5, 0x19, 0x13, 0xcc, 0x1b, 0xfa, 0x34,
-	0x66, 0x89, 0xfe, 0xd3, 0x97, 0x3a, 0x5c, 0x52, 0x92, 0x7d, 0x2b, 0x60, 0x2c, 0x88, 0x68, 0x01,
-	0xf7, 0xfc, 0x24, 0x61, 0xc2, 0x17, 0x21, 0x4b, 0xb8, 0x42, 0xd9, 0x37, 0xb5, 0x55, 0x4a, 0x87,
-	0xf9, 0xc8, 0xa3, 0x71, 0x2a, 0x4e, 0x94, 0xd1, 0xfd, 0xd3, 0x04, 0xd8, 0xce, 0x47, 0x23, 0x9a,
-	0x7d, 0x95, 0x8c, 0x18, 0xda, 0xd0, 0x4e, 0xc3, 0x94, 0x46, 0x61, 0x42, 0x2d, 0xc3, 0x31, 0x7b,
-	0x1d, 0x32, 0x91, 0x71, 0x1d, 0xe0, 0x50, 0x22, 0xbf, 0xf1, 0x63, 0x6a, 0x99, 0xd2, 0x5a, 0xd1,
-	0xa0, 0x0b, 0x6f, 0xa4, 0x34, 0x19, 0x86, 0x49, 0xf0, 0x98, 0xe5, 0x89, 0xb0, 0x5a, 0x8e, 0xd9,
-	0x6b, 0x91, 0x9a, 0x0e, 0x7b, 0x70, 0xd5, 0x1f, 0x1c, 0x1d, 0x54, 0x61, 0x0b, 0x12, 0x36, 0xad,
-	0xc6, 0x0d, 0xe8, 0x0a, 0x26, 0xfc, 0x68, 0x9f, 0x72, 0xee, 0x07, 0x94, 0x5b, 0x8b, 0x12, 0x57,
-	0x57, 0x16, 0x39, 0x15, 0x83, 0x3d, 0x9a, 0x04, 0xe2, 0xb9, 0xb5, 0xa4, 0x72, 0x56, 0x75, 0xb8,
-	0x09, 0xab, 0x4a, 0xfe, 0xae, 0xf0, 0xd9, 0x0b, 0xe3, 0x50, 0x58, 0xcb, 0x8e, 0xd9, 0x33, 0xc8,
-	0x39, 0x3d, 0x3a, 0xb0, 0x52, 0xd1, 0x59, 0x6d, 0x09, 0xab, 0xaa, 0x70, 0x0d, 0x96, 0x42, 0xfe,
-	0x24, 0x8f, 0x22, 0xab, 0xe3, 0x98, 0xbd, 0x36, 0xd1, 0x92, 0xfb, 0xaf, 0x09, 0xdd, 0xef, 0x69,
-	0x26, 0xe8, 0xcf, 0xfb, 0x54, 0x64, 0xe1, 0x80, 0xcf, 0xad, 0xe5, 0x1a, 0x2c, 0x8d, 0x25, 0x58,
-	0xd7, 0x51, 0x4b, 0xf8, 0x2d, 0x5c, 0x4d, 0x33, 0x36, 0xa0, 0x9c, 0x87, 0x49, 0x40, 0x7c, 0x41,
-	0xb9, 0xd5, 0x72, 0x5a, 0xbd, 0x95, 0xad, 0xcd, 0xbe, 0xee, 0x7c, 0x2d, 0x47, 0xff, 0xa0, 0x0e,
-	0xde, 0x49, 0x44, 0x76, 0x42, 0xa6, 0x43, 0xe0, 0x23, 0x68, 0xeb, 0x2e, 0x70, 0x6b, 0x41, 0x86,
-	0xbb, 0x3d, 0x23, 0x9c, 0x46, 0xa9, 0x38, 0x13, 0x27, 0x7b, 0x1b, 0xae, 0x35, 0x65, 0xc2, 0x55,
-	0x68, 0x1d, 0xd1, 0x13, 0xcb, 0x70, 0x8c, 0x5e, 0x87, 0x14, 0x4f, 0xbc, 0x06, 0x8b, 0x63, 0x3f,
-	0xca, 0x8b, 0xf9, 0x30, 0x7a, 0x06, 0x51, 0xc2, 0x03, 0xf3, 0xbe, 0x61, 0x3f, 0x84, 0x6e, 0x2d,
-	0xfc, 0x45, 0xce, 0xad, 0x8a, 0xb3, 0xbb, 0x0d, 0x57, 0x0e, 0x74, 0xed, 0x9e, 0x0a, 0x5f, 0xe4,
-	0xbc, 0xa8, 0x20, 0x97, 0x2f, 0x5d, 0x5b, 0x2d, 0xa1, 0x05, 0xcb, 0xb1, 0x9a, 0x0e, 0x5d, 0xda,
-	0x52, 0x74, 0x3f, 0x01, 0xdc, 0x0b, 0xb9, 0x50, 0xd3, 0xce, 0x09, 0xfd, 0x29, 0xa7, 0x5c, 0xcc,
-	0xeb, 0x92, 0xfb, 0x18, 0xde, 0xaa, 0x79, 0xf0, 0x94, 0x25, 0x9c, 0xe2, 0x5d, 0x58, 0x56, 0x13,
-	0x51, 0xe4, 0x2e, 0xaa, 0x89, 0x65, 0x35, 0x5f, 0x6f, 0x12, 0x29, 0x21, 0xee, 0x13, 0x58, 0xdd,
-	0xa5, 0x3a, 0xc6, 0x25, 0x92, 0x16, 0x1f, 0xa6, 0x5c, 0xcb, 0xd1, 0x50, 0x92, 0xfb, 0x08, 0xde,
-	0xac, 0xc4, 0xd1, 0x54, 0x36, 0x27, 0xe0, 0x22, 0x4c, 0x33, 0x93, 0x32, 0xc0, 0x3d, 0xb0, 0x76,
-	0xa9, 0xa8, 0x97, 0xf1, 0x32, 0x55, 0xf8, 0x1a, 0x6e, 0x34, 0xf8, 0x69, 0x02, 0xfd, 0x5a, 0x1b,
-	0x56, 0xb6, 0xd6, 0x4a, 0x02, 0x53, 0x78, 0x8d, 0x72, 0xf7, 0xe1, 0xed, 0x5d, 0x2a, 0x6a, 0x53,
-	0xd7, 0xc4, 0xc1, 0x9c, 0xb9, 0x2f, 0xad, 0xea, 0xbe, 0xb8, 0xcf, 0xe4, 0x37, 0x4d, 0x85, 0xd3,
-	0xd4, 0x1e, 0x42, 0x77, 0x5c, 0x35, 0xe8, 0x66, 0x5d, 0x6f, 0x1c, 0x7d, 0x52, 0xc7, 0xba, 0xbf,
-	0x1b, 0xd0, 0xdd, 0x19, 0x06, 0xf4, 0x99, 0x2f, 0x68, 0x16, 0xfb, 0xd9, 0xd1, 0xdc, 0x9e, 0x21,
-	0x2c, 0xd0, 0xe1, 0x64, 0xe2, 0xe4, 0xbb, 0x38, 0x97, 0xc7, 0xa5, 0xb3, 0xda, 0xe2, 0x16, 0xa9,
-	0x68, 0xb0, 0x0f, 0x18, 0xf2, 0x49, 0xf8, 0x9d, 0xc4, 0x3f, 0x8c, 0xe8, 0x50, 0x5e, 0xc3, 0x36,
-	0x69, 0xb0, 0x14, 0x39, 0x46, 0x19, 0x8b, 0xe5, 0x1d, 0xec, 0x10, 0xf9, 0xc6, 0x2b, 0x60, 0x0a,
-	0x26, 0x8f, 0x5e, 0x87, 0x98, 0x82, 0xb9, 0x23, 0x78, 0xa7, 0xd2, 0xaa, 0x49, 0x88, 0xd7, 0x35,
-	0xd9, 0x01, 0x4c, 0xcf, 0x59, 0xa7, 0x0b, 0x53, 0xfb, 0x6e, 0xd2, 0xe0, 0xe0, 0x3e, 0x80, 0x5b,
-	0x33, 0xf2, 0x5c, 0x38, 0x4e, 0x5b, 0x7f, 0x2c, 0x42, 0xf7, 0x4b, 0x99, 0xe8, 0x29, 0xcd, 0xc6,
-	0xe1, 0x80, 0xa2, 0x80, 0x95, 0xca, 0x9a, 0xa1, 0x5d, 0xf2, 0x38, 0xbf, 0xad, 0xf6, 0xcd, 0x46,
-	0x9b, 0xfa, 0x38, 0xf7, 0xee, 0x2f, 0xff, 0xfc, 0xff, 0x9b, 0xf9, 0x3e, 0x6e, 0xc8, 0x1f, 0xc2,
-	0xf1, 0xa7, 0x5e, 0x99, 0x93, 0x7b, 0x2f, 0xca, 0xe7, 0xa9, 0xa7, 0xf7, 0x12, 0x8f, 0xa1, 0x33,
-	0xd9, 0x27, 0xb4, 0xca, 0xb8, 0xd3, 0xab, 0x6a, 0xdf, 0x68, 0xb0, 0xe8, 0x7c, 0x9f, 0xc9, 0x7c,
-	0x1e, 0x7e, 0x7c, 0x99, 0x7c, 0xde, 0x0b, 0xf5, 0x38, 0xc5, 0x97, 0x86, 0xbc, 0x08, 0xf5, 0x1f,
-	0x8b, 0x77, 0x2b, 0x69, 0x9a, 0xb6, 0xc3, 0x76, 0x66, 0x03, 0x34, 0x9d, 0xcf, 0x25, 0x9d, 0xfb,
-	0x78, 0x6f, 0x2e, 0x9d, 0x62, 0xcc, 0xc3, 0x41, 0xa1, 0x53, 0x03, 0x7f, 0xea, 0xc5, 0x9a, 0xc2,
-	0x4b, 0x03, 0xae, 0x37, 0x76, 0x15, 0x37, 0x2a, 0xb9, 0x67, 0x36, 0xdd, 0xbe, 0x73, 0x01, 0x4a,
-	0xd3, 0xf4, 0x24, 0xcd, 0x0f, 0xf1, 0x83, 0xb9, 0x34, 0x2b, 0x8b, 0xf2, 0xab, 0x21, 0x2f, 0xdf,
-	0xd4, 0xfd, 0x77, 0x1a, 0xb2, 0xd5, 0x6e, 0x9a, 0xfd, 0xde, 0x1c, 0x84, 0xe6, 0xf2, 0x91, 0xe4,
-	0x72, 0x07, 0x6f, 0xcf, 0xe5, 0xa2, 0x4e, 0xd7, 0xf6, 0x17, 0x7f, 0x9d, 0xad, 0x1b, 0x7f, 0x9f,
-	0xad, 0x1b, 0xff, 0x9d, 0xad, 0x1b, 0x3f, 0x6c, 0x05, 0xa1, 0x78, 0x9e, 0x1f, 0xf6, 0x07, 0x2c,
-	0xf6, 0x92, 0x3c, 0xf6, 0xd3, 0x8c, 0xfd, 0x28, 0x1f, 0xa3, 0x88, 0x1d, 0x7b, 0x8d, 0xff, 0xbc,
-	0xbd, 0x0a, 0x00, 0x00, 0xff, 0xff, 0xc7, 0xe3, 0x7f, 0xc1, 0xd4, 0x09, 0x00, 0x00,
+	0x47, 0x28, 0x67, 0x88, 0x44, 0x55, 0xb5, 0x12, 0x45, 0x29, 0x69, 0x84, 0x48, 0x50, 0xb4, 0x85,
+	0x56, 0xe2, 0xcd, 0xb9, 0xdb, 0x73, 0x4d, 0x6c, 0xaf, 0xf1, 0xae, 0x2f, 0x44, 0x55, 0x5e, 0x90,
+	0xf8, 0x04, 0xa8, 0x9f, 0x04, 0x89, 0x8f, 0x80, 0x78, 0x44, 0xe2, 0x91, 0x17, 0x14, 0xf1, 0x41,
+	0x90, 0x77, 0xd7, 0x57, 0xfb, 0xe2, 0xbb, 0xe4, 0xe9, 0x76, 0x66, 0x7f, 0x33, 0xf3, 0xdb, 0xf9,
+	0xe7, 0x03, 0x37, 0x3d, 0x0a, 0x3c, 0x3f, 0x0d, 0xb9, 0x97, 0x66, 0x4c, 0x30, 0x6f, 0xe8, 0xd3,
+	0x98, 0x25, 0xfa, 0xa7, 0x2f, 0x75, 0xb8, 0xa4, 0x24, 0xfb, 0x56, 0xc0, 0x58, 0x10, 0xd1, 0x02,
+	0xee, 0xf9, 0x49, 0xc2, 0x84, 0x2f, 0x42, 0x96, 0x70, 0x85, 0xb2, 0x6f, 0xea, 0x5b, 0x29, 0x1d,
+	0xe6, 0x23, 0x8f, 0xc6, 0xa9, 0x38, 0x51, 0x97, 0xee, 0x1f, 0x26, 0xc0, 0x76, 0x3e, 0x1a, 0xd1,
+	0xec, 0xab, 0x64, 0xc4, 0xd0, 0x86, 0x76, 0x1a, 0xa6, 0x34, 0x0a, 0x13, 0x6a, 0x19, 0x8e, 0xd9,
+	0xeb, 0x90, 0x89, 0x8c, 0xeb, 0x00, 0x87, 0x12, 0xf9, 0x8d, 0x1f, 0x53, 0xcb, 0x94, 0xb7, 0x15,
+	0x0d, 0xba, 0xf0, 0x46, 0x4a, 0x93, 0x61, 0x98, 0x04, 0x8f, 0x59, 0x9e, 0x08, 0xab, 0xe5, 0x98,
+	0xbd, 0x16, 0xa9, 0xe9, 0xb0, 0x07, 0x57, 0xfd, 0xc1, 0xd1, 0x41, 0x15, 0xb6, 0x20, 0x61, 0xd3,
+	0x6a, 0xdc, 0x80, 0xae, 0x60, 0xc2, 0x8f, 0xf6, 0x29, 0xe7, 0x7e, 0x40, 0xb9, 0xb5, 0x28, 0x71,
+	0x75, 0x65, 0x11, 0x53, 0x31, 0xd8, 0xa3, 0x49, 0x20, 0x5e, 0x58, 0x4b, 0x2a, 0x66, 0x55, 0x87,
+	0x9b, 0xb0, 0xaa, 0xe4, 0xef, 0x0a, 0x9b, 0xbd, 0x30, 0x0e, 0x85, 0xb5, 0xec, 0x98, 0x3d, 0x83,
+	0x9c, 0xd3, 0xa3, 0x03, 0x2b, 0x15, 0x9d, 0xd5, 0x96, 0xb0, 0xaa, 0x0a, 0xd7, 0x60, 0x29, 0xe4,
+	0x4f, 0xf2, 0x28, 0xb2, 0x3a, 0x8e, 0xd9, 0x6b, 0x13, 0x2d, 0xb9, 0xff, 0x98, 0xd0, 0x7d, 0x46,
+	0x33, 0x41, 0x7f, 0xda, 0xa7, 0x22, 0x0b, 0x07, 0x7c, 0x6e, 0x2e, 0xd7, 0x60, 0x69, 0x2c, 0xc1,
+	0x3a, 0x8f, 0x5a, 0xc2, 0x6f, 0xe1, 0x6a, 0x9a, 0xb1, 0x01, 0xe5, 0x3c, 0x4c, 0x02, 0xe2, 0x0b,
+	0xca, 0xad, 0x96, 0xd3, 0xea, 0xad, 0x6c, 0x6d, 0xf6, 0x75, 0xe5, 0x6b, 0x31, 0xfa, 0x07, 0x75,
+	0xf0, 0x4e, 0x22, 0xb2, 0x13, 0x32, 0xed, 0x02, 0x1f, 0x41, 0x5b, 0x57, 0x81, 0x5b, 0x0b, 0xd2,
+	0xdd, 0xed, 0x19, 0xee, 0x34, 0x4a, 0xf9, 0x99, 0x18, 0xd9, 0xdb, 0x70, 0xad, 0x29, 0x12, 0xae,
+	0x42, 0xeb, 0x88, 0x9e, 0x58, 0x86, 0x63, 0xf4, 0x3a, 0xa4, 0x38, 0xe2, 0x35, 0x58, 0x1c, 0xfb,
+	0x51, 0x5e, 0xf4, 0x87, 0xd1, 0x33, 0x88, 0x12, 0x1e, 0x98, 0xf7, 0x0d, 0xfb, 0x21, 0x74, 0x6b,
+	0xee, 0x2f, 0x32, 0x6e, 0x55, 0x8c, 0xdd, 0x67, 0x70, 0xe5, 0x40, 0xe7, 0xee, 0xa9, 0xf0, 0x45,
+	0xce, 0x8b, 0x0c, 0x72, 0x79, 0xd2, 0xb9, 0xd5, 0x12, 0x5a, 0xb0, 0x1c, 0xab, 0xee, 0xd0, 0xa9,
+	0x2d, 0x45, 0x44, 0x58, 0x18, 0xb0, 0x21, 0x95, 0x7d, 0xd9, 0x21, 0xf2, 0xec, 0x7e, 0x02, 0xb8,
+	0x17, 0x72, 0xa1, 0x26, 0x80, 0x13, 0xfa, 0x63, 0x4e, 0xb9, 0x98, 0x57, 0x39, 0xf7, 0x31, 0xbc,
+	0x55, 0xb3, 0xe0, 0x29, 0x4b, 0x38, 0xc5, 0xbb, 0xb0, 0xac, 0xba, 0xa4, 0xe0, 0x53, 0x64, 0x18,
+	0xcb, 0x0c, 0xbf, 0x9e, 0x2e, 0x52, 0x42, 0xdc, 0x27, 0xb0, 0xba, 0x4b, 0xb5, 0x8f, 0x4b, 0x04,
+	0x2d, 0x1e, 0xab, 0x4c, 0xcb, 0x76, 0x51, 0x92, 0xfb, 0x08, 0xde, 0xac, 0xf8, 0xd1, 0x54, 0x36,
+	0x27, 0xe0, 0xc2, 0x4d, 0x33, 0x93, 0xd2, 0xc1, 0x3d, 0xb0, 0x76, 0xa9, 0xa8, 0xa7, 0xf6, 0x32,
+	0x59, 0xf8, 0x1a, 0x6e, 0x34, 0xd8, 0x69, 0x02, 0xfd, 0x5a, 0x69, 0x56, 0xb6, 0xd6, 0x4a, 0x02,
+	0x53, 0x78, 0x8d, 0x72, 0xf7, 0xe1, 0xed, 0x5d, 0x2a, 0x6a, 0x9d, 0xd8, 0xc4, 0xc1, 0x9c, 0x39,
+	0x43, 0xad, 0xea, 0x0c, 0xb9, 0xcf, 0xe5, 0x9b, 0xa6, 0xdc, 0x69, 0x6a, 0x0f, 0xa1, 0x3b, 0xae,
+	0x5e, 0xe8, 0x62, 0x5d, 0x6f, 0x1c, 0x07, 0x52, 0xc7, 0xba, 0xbf, 0x19, 0xd0, 0xdd, 0x19, 0x06,
+	0xf4, 0xb9, 0x2f, 0x68, 0x16, 0xfb, 0xd9, 0xd1, 0xdc, 0x9a, 0x21, 0x2c, 0xd0, 0xe1, 0xa4, 0x0b,
+	0xe5, 0xb9, 0x58, 0xa1, 0xc7, 0xa5, 0xb1, 0x9a, 0xec, 0x16, 0xa9, 0x68, 0xb0, 0x0f, 0x18, 0xf2,
+	0x89, 0xfb, 0x9d, 0xc4, 0x3f, 0x8c, 0xe8, 0x50, 0x6e, 0xc8, 0x36, 0x69, 0xb8, 0x29, 0x62, 0x8c,
+	0x32, 0x16, 0xcb, 0xdd, 0xd8, 0x21, 0xf2, 0x8c, 0x57, 0xc0, 0x14, 0x4c, 0x2e, 0xc2, 0x0e, 0x31,
+	0x05, 0x73, 0x47, 0xf0, 0x4e, 0xa5, 0x54, 0x13, 0x17, 0xaf, 0x73, 0xb2, 0x03, 0x98, 0x9e, 0xbb,
+	0x9d, 0x4e, 0x4c, 0xed, 0xdd, 0xa4, 0xc1, 0xc0, 0x7d, 0x00, 0xb7, 0x66, 0xc4, 0xb9, 0xb0, 0x9d,
+	0xb6, 0x7e, 0x5f, 0x84, 0xee, 0x97, 0x32, 0xd0, 0x53, 0x9a, 0x8d, 0xc3, 0x01, 0x45, 0x01, 0x2b,
+	0x95, 0x31, 0x43, 0xbb, 0xe4, 0x71, 0x7e, 0x5a, 0xed, 0x9b, 0x8d, 0x77, 0xea, 0x71, 0xee, 0xdd,
+	0x9f, 0xff, 0xfe, 0xef, 0x57, 0xf3, 0x7d, 0xdc, 0x90, 0x1f, 0xc7, 0xf1, 0xa7, 0x5e, 0x19, 0x93,
+	0x7b, 0x2f, 0xcb, 0xe3, 0xa9, 0xa7, 0xe7, 0x12, 0x8f, 0xa1, 0x33, 0x99, 0x27, 0xb4, 0x4a, 0xbf,
+	0xd3, 0xa3, 0x6a, 0xdf, 0x68, 0xb8, 0xd1, 0xf1, 0x3e, 0x93, 0xf1, 0x3c, 0xfc, 0xf8, 0x32, 0xf1,
+	0xbc, 0x97, 0xea, 0x70, 0x8a, 0xaf, 0x0c, 0xb9, 0x11, 0xea, 0x1f, 0x90, 0x77, 0x2b, 0x61, 0x9a,
+	0xa6, 0xc3, 0x76, 0x66, 0x03, 0x34, 0x9d, 0xcf, 0x25, 0x9d, 0xfb, 0x78, 0x6f, 0x2e, 0x9d, 0xa2,
+	0xcd, 0xc3, 0x41, 0xa1, 0x53, 0x0d, 0x7f, 0xea, 0xc5, 0x9a, 0xc2, 0x2b, 0x03, 0xae, 0x37, 0x56,
+	0x15, 0x37, 0x2a, 0xb1, 0x67, 0x16, 0xdd, 0xbe, 0x73, 0x01, 0x4a, 0xd3, 0xf4, 0x24, 0xcd, 0x0f,
+	0xf1, 0x83, 0xb9, 0x34, 0x2b, 0x83, 0xf2, 0x8b, 0x21, 0x37, 0xdf, 0xd4, 0x37, 0xc1, 0x69, 0x88,
+	0x56, 0xdb, 0x69, 0xf6, 0x7b, 0x73, 0x10, 0x9a, 0xcb, 0x47, 0x92, 0xcb, 0x1d, 0xbc, 0x3d, 0x97,
+	0x8b, 0x5a, 0x5d, 0xdb, 0x5f, 0xfc, 0x79, 0xb6, 0x6e, 0xfc, 0x75, 0xb6, 0x6e, 0xfc, 0x7b, 0xb6,
+	0x6e, 0x7c, 0xbf, 0x15, 0x84, 0xe2, 0x45, 0x7e, 0xd8, 0x1f, 0xb0, 0xd8, 0x4b, 0xf2, 0xd8, 0x4f,
+	0x33, 0xf6, 0x83, 0x3c, 0x8c, 0x22, 0x76, 0xec, 0x35, 0xfe, 0xa1, 0xfb, 0x3f, 0x00, 0x00, 0xff,
+	0xff, 0x85, 0xa5, 0xb2, 0x62, 0xe8, 0x09, 0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -1363,6 +1372,15 @@ func (m *PipelineStatus) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
+	if m.Code == nil {
+		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("code")
+	} else {
+		i -= len(*m.Code)
+		copy(dAtA[i:], *m.Code)
+		i = encodeVarintDaemon(dAtA, i, uint64(len(*m.Code)))
+		i--
+		dAtA[i] = 0x1a
+	}
 	if m.Message == nil {
 		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("message")
 	} else {
@@ -1971,6 +1989,10 @@ func (m *PipelineStatus) Size() (n int) {
 		l = len(*m.Message)
 		n += 1 + l + sovDaemon(uint64(l))
 	}
+	if m.Code != nil {
+		l = len(*m.Code)
+		n += 1 + l + sovDaemon(uint64(l))
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
@@ -2915,6 +2937,40 @@ func (m *PipelineStatus) Unmarshal(dAtA []byte) error {
 			m.Message = &s
 			iNdEx = postIndex
 			hasFields[0] |= uint64(0x00000002)
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Code", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowDaemon
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			s := string(dAtA[iNdEx:postIndex])
+			m.Code = &s
+			iNdEx = postIndex
+			hasFields[0] |= uint64(0x00000004)
 		default:
 			iNdEx = preIndex
 			skippy, err := skipDaemon(dAtA[iNdEx:])
@@ -2937,6 +2993,9 @@ func (m *PipelineStatus) Unmarshal(dAtA []byte) error {
 	if hasFields[0]&uint64(0x00000002) == 0 {
 		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("message")
 	}
+	if hasFields[0]&uint64(0x00000004) == 0 {
+		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("code")
+	}
 
 	if iNdEx > l {
 		return io.ErrUnexpectedEOF
diff --git a/pkg/apis/proto/daemon/daemon.proto b/pkg/apis/proto/daemon/daemon.proto
index 56c8191a..87a6786a 100644
--- a/pkg/apis/proto/daemon/daemon.proto
+++ b/pkg/apis/proto/daemon/daemon.proto
@@ -49,6 +49,7 @@ message VertexMetrics {
 message PipelineStatus {
   required string status = 1;
   required string message = 2;
+  required string code = 3;
 }
 
 message ListBuffersRequest {
diff --git a/pkg/daemon/server/daemon_server.go b/pkg/daemon/server/daemon_server.go
index 6ba93b6d..a958c36d 100644
--- a/pkg/daemon/server/daemon_server.go
+++ b/pkg/daemon/server/daemon_server.go
@@ -48,14 +48,16 @@ import (
 )
 
 type daemonServer struct {
-	pipeline   *v1alpha1.Pipeline
-	isbSvcType v1alpha1.ISBSvcType
+	pipeline      *v1alpha1.Pipeline
+	isbSvcType    v1alpha1.ISBSvcType
+	metaDataQuery *service.PipelineMetadataQuery
 }
 
 func NewDaemonServer(pl *v1alpha1.Pipeline, isbSvcType v1alpha1.ISBSvcType) *daemonServer {
 	return &daemonServer{
-		pipeline:   pl,
-		isbSvcType: isbSvcType,
+		pipeline:      pl,
+		isbSvcType:    isbSvcType,
+		metaDataQuery: nil,
 	}
 }
 
@@ -138,11 +140,17 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 	go func() { _ = httpServer.Serve(httpL) }()
 	go func() { _ = tcpm.Serve() }()
 
+	// Start the Data flow health status updater
+	go func() {
+		ds.metaDataQuery.StartHealthCheck(ctx)
+	}()
+
 	log.Infof("Daemon server started successfully on %s", address)
 	// Start the rater
 	if err := rater.Start(ctx); err != nil {
 		return fmt.Errorf("failed to start the rater: %w", err)
 	}
+
 	<-ctx.Done()
 	return nil
 }
@@ -169,6 +177,7 @@ func (ds *daemonServer) newGRPCServer(
 		return nil, err
 	}
 	daemon.RegisterDaemonServiceServer(grpcServer, pipelineMetadataQuery)
+	ds.metaDataQuery = pipelineMetadataQuery
 	return grpcServer, nil
 }
 
diff --git a/pkg/daemon/server/service/healthStatus.go b/pkg/daemon/server/service/healthStatus.go
new file mode 100644
index 00000000..47b1e113
--- /dev/null
+++ b/pkg/daemon/server/service/healthStatus.go
@@ -0,0 +1,462 @@
+package service
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+	"golang.org/x/net/context"
+
+	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/apis/proto/daemon"
+	"github.com/numaproj/numaflow/pkg/isbsvc"
+	"github.com/numaproj/numaflow/pkg/shared/ewma"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
+)
+
+const (
+	// healthSlidingWindow is the total time window to compute the health of a vertex
+	healthSlidingWindow = 5 * time.Minute
+
+	// healthTimeStep is the frequency at which the health of a vertex is computed
+	healthTimeStep = 10 * time.Second
+
+	// healthWindowSize is the number of timeline entries to keep for a given vertex
+	// This is used to compute the health of a vertex
+	// Ensure that healthSlidingWindow / healthTimeStep is an integer
+	healthWindowSize = healthSlidingWindow / healthTimeStep
+
+	// criticalWindowTime is the number of entries to look back to assign a critical state to a vertex
+	// Critical state is assigned to a vertex if the weighted mean average buffer usage of
+	// the vertex for a given time period is above a defined criticalBufferThreshold
+	criticalWindowTime = 1 * time.Minute
+
+	// criticalWindowSize is the number of entries to look back to assign a critical state to a vertex
+	// This is used to avoid false positives
+	criticalWindowSize = int(criticalWindowTime / healthTimeStep)
+
+	// criticalLookBackCount is the number of times the state must be critical in the last criticalWindowSize entries
+	// This is used to avoid false positives
+	criticalLookBackCount = 3
+)
+
+// HealthThresholds are the thresholds used to compute the health of a vertex
+const (
+	// criticalBufferThreshold is the threshold above which the health of a vertex is critical
+	criticalBufferThreshold = 95
+	// warningBufferThreshold is the threshold above which the health of a vertex is warning
+	warningBufferThreshold = 80
+)
+
+// Dataflow states
+const (
+	// criticalState is the state of a vertex when its health is critical
+	criticalState = "critical"
+	// warningState is the state of a vertex when its health is warning
+	warningState = "warning"
+	// healthyState is the state of a vertex when its health is healthy
+	healthyState = "healthy"
+)
+
+const (
+	// enableCriticalLookBack is used to enable the look back for critical state
+	enableCriticalLookBack = true
+)
+
+// timelineEntry is used to store the buffer usage timeline for a given vertex
+type timelineEntry struct {
+	// The time at which the entry is recorded
+	Time int64 `json:"time"`
+	// The buffer usage of the pipeline at the time
+	BufferUsage float64 `json:"bufferUsage"`
+	// The rolling average buffer usage of the pipeline over the last HEALTH_WINDOW_SIZE seconds
+	AverageBufferUsage float64 `json:"averageBufferUsage"`
+}
+
+// dataHealthResponse is the response returned by the data health check API
+type dataHealthResponse struct {
+	// Status is the overall data status of the pipeline
+	Status string `json:"status"`
+	// Message is the error message if any
+	Message string `json:"message"`
+	// Code is the status code for the data health
+	Code string `json:"code"`
+}
+
+// newDataHealthResponse is used to create a new dataHealthResponse object
+func newDataHealthResponse(status string, message string, code string) *dataHealthResponse {
+	return &dataHealthResponse{
+		Status:  status,
+		Message: message,
+		Code:    code,
+	}
+}
+
+// defaultDataHealthResponse is the default response returned by the data health check API
+var defaultDataHealthResponse = newDataHealthResponse(PipelineStatusUnknown,
+	"Pipeline dataflow is in an unknown state",
+	"D4")
+
+// HealthChecker is the struct type for health checker.
+type HealthChecker struct {
+	// Add a field for the health status.
+	currentDataStatus *dataHealthResponse
+	isbSvcClient      isbsvc.ISBService
+	pipeline          *v1alpha1.Pipeline
+	timelineData      map[string]*sharedqueue.OverflowQueue[*timelineEntry]
+	statusLock        *sync.RWMutex
+}
+
+// NewHealthChecker creates a new object HealthChecker struct type.
+func NewHealthChecker(pipeline *v1alpha1.Pipeline, isbSvcClient isbsvc.ISBService) *HealthChecker {
+	// Return a new HealthChecker struct instance.
+	return &HealthChecker{
+		currentDataStatus: defaultDataHealthResponse,
+		isbSvcClient:      isbSvcClient,
+		pipeline:          pipeline,
+		timelineData:      make(map[string]*sharedqueue.OverflowQueue[*timelineEntry]),
+		statusLock:        &sync.RWMutex{},
+	}
+}
+
+// vertexState is a struct which contains the name and  state of a vertex
+type vertexState struct {
+	// Name is the name of the vertex
+	Name string `json:"name"`
+	// State is the state of the vertex
+	State string `json:"state"`
+}
+
+// newVertexState is used to create a new vertexState object
+func newVertexState(name string, state string) *vertexState {
+	return &vertexState{
+		Name:  name,
+		State: state,
+	}
+}
+
+// getCurrentHealth returns the current health status of the pipeline.
+// It is thread safe to ensure concurrent access.
+func (hc *HealthChecker) getCurrentHealth() *dataHealthResponse {
+	// Lock the statusLock to ensure thread safety.
+	hc.statusLock.RLock()
+	defer hc.statusLock.RUnlock()
+	// Return the health status.
+	return hc.currentDataStatus
+}
+
+// setCurrentHealth sets the current health status of the pipeline.
+// It is thread safe to ensure concurrent access.
+func (hc *HealthChecker) setCurrentHealth(status *dataHealthResponse) {
+	// Lock the statusLock to ensure thread safety.
+	hc.statusLock.Lock()
+	defer hc.statusLock.Unlock()
+	// Set the health status.
+	hc.currentDataStatus = status
+}
+
+// startHealthCheck starts the health check for the pipeline.
+// The ticks are generated at the interval of healthTimeStep.
+func (hc *HealthChecker) startHealthCheck(ctx context.Context) {
+	logger := logging.FromContext(ctx)
+	// Goroutine to listen for ticks
+	// At every tick, check and update the health status of the pipeline.
+	// If the context is done, return.
+	// Create a ticker to generate ticks at the interval of healthTimeStep.
+	ticker := time.NewTicker(healthTimeStep)
+	defer ticker.Stop()
+	for {
+		select {
+		// If the ticker ticks, check and update the health status of the pipeline.
+		case <-ticker.C:
+			// Get the current health status of the pipeline.
+			criticality, err := hc.getPipelineVertexDataCriticality(ctx)
+			logger.Debugw("Health check", zap.Any("criticality", criticality))
+			if err != nil {
+				// If there is an error, set the current health status to unknown.
+				// as we are not able to determine the health of the pipeline.
+				logger.Errorw("Failed to vertex data criticality", zap.Error(err))
+				hc.setCurrentHealth(defaultDataHealthResponse)
+			} else {
+				// convert the vertex state to pipeline state
+				pipelineState := convertVertexStateToPipelineState(criticality)
+				// update the current health status of the pipeline
+				hc.setCurrentHealth(pipelineState)
+			}
+		// If the context is done, return.
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// getPipelineVertexDataCriticality is used to provide the data criticality of the pipeline
+// They can be of the following types:
+// 1. Ok: The pipeline is working as expected
+// 2. Warning: The pipeline is working but there is a lag in the data movement
+// 3. Critical: The pipeline is not working as expected
+// We need to check the following things to determine the data criticality of the pipeline:
+// 1. The buffer usage of each buffer in the pipeline
+//
+// Based on this information, we first get the current buffer usage of each buffer in the pipeline at that instant
+// and populate the timeline data for each buffer.
+// Then for a given buffer, we calculate the weighted mean average buffer usage with decay over
+// the last HEALTH_WINDOW_SIZE seconds. This is done to smoothen the curve and remove point spikes
+// If this average is greater than the critical threshold, we return the critical status
+// If this average is greater than the warning threshold, we return the warning status
+// If this average is less than the warning threshold, we return the ok status
+func (hc *HealthChecker) getPipelineVertexDataCriticality(ctx context.Context) ([]*vertexState, error) {
+	// Fetch the buffer information for the pipeline
+	buffers, err := listBuffers(ctx, hc.pipeline, hc.isbSvcClient)
+	if err != nil {
+		return nil, err
+	}
+	// update the usage timeline for all the ISBs used in the pipeline
+	hc.updateUsageTimeline(buffers.Buffers)
+
+	var vertexState []*vertexState
+
+	// iterate over the timeline data for each buffer and calculate the exponential weighted mean average
+	// for the last HEALTH_WINDOW_SIZE buffer usage entries
+	for bufferName := range hc.timelineData {
+		// Extract the buffer usage of the timeline
+		var bufferUsage []float64
+		for _, entry := range hc.timelineData[bufferName].Items() {
+			bufferUsage = append(bufferUsage, entry.BufferUsage)
+		}
+		// calculate the average buffer usage over the last HEALTH_WINDOW_SIZE seconds
+		ewmaBufferUsage := calculateEWMAUsage(bufferUsage)
+		// assign the state to the vertex based on the average buffer usage
+		// Look back is enabled for the critical state
+		currentState := assignStateToTimeline(ewmaBufferUsage, enableCriticalLookBack)
+		// create a new vertex state object
+		currentVertexState := newVertexState(bufferName, currentState)
+		// add the vertex state to the list of vertex states
+		vertexState = append(vertexState, currentVertexState)
+	}
+	return vertexState, nil
+}
+
+// updateUsageTimeline is used to update the usage timeline for a given buffer list
+// This iterates over all the buffers in the buffer list and updates the usage timeline for each buffer
+// The timeline data is represented as a map of buffer name to a list of timelineEntry
+// Example:
+//
+//	{
+//		"bufferName": [
+//			{
+//				"time": 1234567890,
+//				"bufferUsage": 0.5,
+//				"averageBufferUsage": 0.5
+//			},
+func (hc *HealthChecker) updateUsageTimeline(bufferList []*daemon.BufferInfo) {
+	// iterate over the buffer list and update the usage timeline for each buffer
+	for _, buffer := range bufferList {
+		// get the buffer name and the current timestamp
+		bufferName := buffer.GetBufferName()
+		timestamp := time.Now().Unix()
+
+		// if the buffer name is not present in the timeline data, add it
+		if _, ok := hc.timelineData[bufferName]; !ok {
+			hc.timelineData[bufferName] = sharedqueue.New[*timelineEntry](int(healthWindowSize))
+		}
+		// extract the current buffer usage and update the average buffer usage
+		bufferUsage := buffer.GetBufferUsage() * 100
+		newAverage := updateAverageBufferUsage(hc.timelineData[bufferName].Items(), bufferUsage)
+		// add the new entry to the timeline
+		hc.timelineData[bufferName].Append(&timelineEntry{
+			Time:               timestamp,
+			BufferUsage:        bufferUsage,
+			AverageBufferUsage: newAverage,
+		})
+	}
+}
+
+// updateAverageBufferUsage computes the average buffer usage of a vertex over the last HEALTH_WINDOW_SIZE seconds
+// This is a rolling average, so the oldest entry is removed from the timeline and the new entry is added
+// The average buffer usage is computed as the average of the buffer usage of the last HEALTH_WINDOW_SIZE entries
+// If the timeline is empty, the average buffer usage is set to the buffer usage of the new entry
+// If the timeline is not empty, the average buffer usage is computed as follows:
+//
+//	averageBufferUsage = lastAverage*HEALTH_WINDOW_SIZE + (newUsageEntry - oldestEntry) / HEALTH_WINDOW_SIZE
+//	where oldestEntry is the oldest entry in the timeline
+//	and lastAverage is the average buffer usage of the last entry in the timeline
+//
+// If the timeline size is less than HEALTH_WINDOW_SIZE, the average buffer usage is computed as follows:
+//
+//	averageBufferUsage = lastAverage * timelineSize + (newUsageEntry) / timelineSize + 1
+//	where lastAverage is the average buffer usage of the last entry in the timeline
+//	and timelineSize is the size of the timeline
+func updateAverageBufferUsage(timeline []*timelineEntry, newEntry float64) float64 {
+	// If the timeline is empty, return the buffer usage of the new entry
+	if len(timeline) == 0 {
+		return newEntry
+	}
+	// Compute the current average buffer usage of the last entry
+	lastAverage := timeline[len(timeline)-1].AverageBufferUsage
+	var newAverage float64
+	// If the timeline is full
+	if len(timeline) == int(healthWindowSize) {
+		newAverage = (lastAverage*float64(healthWindowSize) + newEntry - timeline[0].BufferUsage) / float64(healthWindowSize)
+	} else if len(timeline) < int(healthWindowSize) {
+		// If the timeline is not full
+		newAverage = (lastAverage*float64(len(timeline)) + newEntry) / float64(len(timeline)+1)
+	}
+	return newAverage
+}
+
+// calculateEWMAUsage computes the EWMA buffer usage of a vertex over the last HEALTH_WINDOW_SIZE seconds
+// This is an exponentially weighted moving average, with the following parameters:
+// alpha = 2 / (HEALTH_WINDOW_SIZE + 1)
+// EWMA = alpha * newUsageEntry + (1 - alpha) * lastEWMA
+// where lastEWMA is the EWMA buffer usage of the last entry in the timeline
+func calculateEWMAUsage(bufferUsage []float64) []float64 {
+	// Compute the current EWMA buffer usage of the timeline
+	a := ewma.NewSimpleEWMA(float64(healthWindowSize))
+	var emwaValues []float64
+	// TODO: Check if we can keep storing the EWMA values instead of recomputing them
+	for _, f := range bufferUsage {
+		a.Add(f)
+		emwaValues = append(emwaValues, a.Get())
+	}
+	return emwaValues
+}
+
+// assignStateToBufferUsage assigns a state to the buffer usage of a vertex
+// The state is assigned as follows:
+// - if the buffer usage is above CRITICAL_THRESHOLD, the state is set to CRITICAL
+// - if the buffer usage is above WARNING_THRESHOLD, the state is set to WARNING
+// - otherwise, the state is set to HEALTHY
+func assignStateToBufferUsage(ewmaValue float64) string {
+	// Assign the state to the buffer usage
+	var state string
+	if ewmaValue > criticalBufferThreshold {
+		state = criticalState
+	} else if ewmaValue > warningBufferThreshold {
+		state = warningState
+	} else {
+		state = healthyState
+	}
+	return state
+}
+
+// assignStateToTimeline assigns a state to the timeline of a vertex
+// We also have a look back for critical state to avoid false positives
+// In this case, we check if the state is CRITICAL at least LOOK_BACK_COUNT times in the last CRITICAL_WINDOW_SIZE entries
+// If the state is CRITICAL at least LOOK_BACK_COUNT times in the last CRITICAL_WINDOW_SIZE entries
+// Set the state to CRITICAL
+func assignStateToTimeline(ewmaValues []float64, lookBack bool) string {
+	// Extract the last entry of the timeline
+	ewmaUsage := ewmaValues[len(ewmaValues)-1]
+
+	// Assign the state to the buffer usage value
+	state := assignStateToBufferUsage(ewmaUsage)
+
+	// If the state is CRITICAL, and we have a look back, we need to check we have
+	// LOOK_BACK_COUNT entries as CRITICAL
+	if state == criticalState && lookBack {
+		// Extract the states of the timeline
+		var states []string
+		for _, entry := range ewmaValues {
+			states = append(states, assignStateToBufferUsage(entry))
+		}
+		// Count the number of times the state is CRITICAL in the last CRITICAL_WINDOW_SIZE entries
+		var criticalCount int
+		for i := len(states) - 1; i >= 0; i-- {
+			if states[i] == criticalState {
+				criticalCount++
+			}
+			if (len(states) - i) == criticalWindowSize {
+				break
+			}
+		}
+		// If the state is CRITICAL at least LOOK_BACK_COUNT times in the last CRITICAL_WINDOW_SIZE entries
+		// Set the state to CRITICAL, otherwise set the state to WARNING
+		if criticalCount >= criticalLookBackCount {
+			state = criticalState
+		} else {
+			state = warningState
+		}
+	}
+	return state
+}
+
+// convertVertexStateToPipelineState is used to convert the vertex state to pipeline state
+// if any of the vertices are critical, the pipeline is critical
+// if any of the vertices are warning, the pipeline is warning
+// if any the vertices are unknown, the pipeline is unknown
+// else the pipeline is ok
+// Here we follow a precedence order of unknown > critical > warning  > ok
+// Hence, whichever is the highest precedence state found in the vertex state, we return that
+func convertVertexStateToPipelineState(vertexState []*vertexState) *dataHealthResponse {
+	// create a map to store the precedence order of the states
+	// assign a number to each state based on the precedence order
+
+	// unknown > critical > warning > ok
+	stateMap := map[string]int{
+		PipelineStatusUnknown:  3,
+		PipelineStatusCritical: 2,
+		PipelineStatusWarning:  1,
+		PipelineStatusOK:       0,
+	}
+
+	// initialize the max state to 0 (ie Healthy state)
+	maxState := 0
+	// initialize the max state vertex to empty string
+	maxStateVtx := ""
+
+	// iterate over the vertex state and assign a number to each state and update the current max state
+	for _, state := range vertexState {
+		if stateMap[state.State] > maxState {
+			maxState = stateMap[state.State]
+			maxStateVtx = state.Name
+		}
+	}
+
+	// get the state and vertex corresponding to the max state
+	for state, value := range stateMap {
+		if value == maxState {
+			return generateDataHealthResponse(state, maxStateVtx)
+		}
+	}
+
+	// if we reach here, return unknown state
+	return newDataHealthResponse(PipelineStatusUnknown,
+		"Pipeline dataflow is in an unknown state",
+		"D4")
+}
+
+// generateDataHealthResponse is used to generate the data health response
+// if the state is Healthy we return a message saying the pipeline is healthy, and the code corresponding to Healthy
+// if the state is Warning we return a message saying the pipeline is warning due to the vertex,
+// and the code corresponding to Warning, similar for Critical
+// if the state is Unknown we return a message saying the pipeline is in an unknown state due to the vertex,
+// and the code corresponding to Unknown
+func generateDataHealthResponse(state string, vertex string) *dataHealthResponse {
+	switch state {
+	case PipelineStatusOK:
+		return newDataHealthResponse(
+			PipelineStatusOK,
+			"Pipeline dataflow is healthy",
+			"D1")
+	case PipelineStatusWarning:
+		return newDataHealthResponse(
+			PipelineStatusWarning,
+			fmt.Sprintf("Dataflow is in warning state for  %s", vertex),
+			"D2")
+	case PipelineStatusCritical:
+		return newDataHealthResponse(
+			PipelineStatusCritical,
+			fmt.Sprintf("Dataflow is in critical state for %s", vertex),
+			"D3")
+	case PipelineStatusUnknown:
+		return newDataHealthResponse(
+			PipelineStatusUnknown,
+			fmt.Sprintf("Pipeline dataflow is in an unknown state due to %s", vertex),
+			"D4")
+	default:
+		return defaultDataHealthResponse
+	}
+}
diff --git a/pkg/daemon/server/service/pipeline_metrics_query.go b/pkg/daemon/server/service/pipeline_metrics_query.go
index ee88b76d..f2fa316a 100644
--- a/pkg/daemon/server/service/pipeline_metrics_query.go
+++ b/pkg/daemon/server/service/pipeline_metrics_query.go
@@ -44,19 +44,24 @@ type metricsHttpClient interface {
 	Get(url string) (*http.Response, error)
 }
 
-// pipelineMetadataQuery has the metadata required for the pipeline queries
-type pipelineMetadataQuery struct {
+// PipelineMetadataQuery has the metadata required for the pipeline queries
+type PipelineMetadataQuery struct {
 	isbSvcClient      isbsvc.ISBService
 	pipeline          *v1alpha1.Pipeline
 	httpClient        metricsHttpClient
 	watermarkFetchers map[v1alpha1.Edge][]fetch.HeadFetcher
 	rater             rater.Ratable
+	healthChecker     *HealthChecker
 }
 
 const (
-	PipelineStatusOK      = "OK"
-	PipelineStatusError   = "Error"
-	PipelineStatusUnknown = "Unknown"
+	PipelineStatusOK       = "OK"
+	PipelineStatusError    = "Error"
+	PipelineStatusUnknown  = "unknown"
+	PipelineStatusCritical = "critical"
+	PipelineStatusWarning  = "warning"
+	PipelineStatusInactive = "inactive"
+	PipelineStatusDeleting = "deleting"
 )
 
 // NewPipelineMetadataQuery returns a new instance of pipelineMetadataQuery
@@ -64,9 +69,9 @@ func NewPipelineMetadataQuery(
 	isbSvcClient isbsvc.ISBService,
 	pipeline *v1alpha1.Pipeline,
 	wmFetchers map[v1alpha1.Edge][]fetch.HeadFetcher,
-	rater rater.Ratable) (*pipelineMetadataQuery, error) {
+	rater rater.Ratable) (*PipelineMetadataQuery, error) {
 	var err error
-	ps := pipelineMetadataQuery{
+	ps := PipelineMetadataQuery{
 		isbSvcClient: isbSvcClient,
 		pipeline:     pipeline,
 		httpClient: &http.Client{
@@ -77,6 +82,7 @@ func NewPipelineMetadataQuery(
 		},
 		watermarkFetchers: wmFetchers,
 		rater:             rater,
+		healthChecker:     NewHealthChecker(pipeline, isbSvcClient),
 	}
 	if err != nil {
 		return nil, err
@@ -85,45 +91,16 @@ func NewPipelineMetadataQuery(
 }
 
 // ListBuffers is used to obtain the all the edge buffers information of a pipeline
-func (ps *pipelineMetadataQuery) ListBuffers(ctx context.Context, req *daemon.ListBuffersRequest) (*daemon.ListBuffersResponse, error) {
-	log := logging.FromContext(ctx)
-	resp := new(daemon.ListBuffersResponse)
-
-	buffers := []*daemon.BufferInfo{}
-	for _, buffer := range ps.pipeline.GetAllBuffers() {
-		bufferInfo, err := ps.isbSvcClient.GetBufferInfo(ctx, buffer)
-		if err != nil {
-			return nil, fmt.Errorf("failed to get information of buffer %q", buffer)
-		}
-		log.Debugf("Buffer %s has bufferInfo %+v", buffer, bufferInfo)
-		v := ps.pipeline.FindVertexWithBuffer(buffer)
-		if v == nil {
-			return nil, fmt.Errorf("unexpected error, buffer %q not found from the pipeline", buffer)
-		}
-		bufferLength, bufferUsageLimit := getBufferLimits(ps.pipeline, *v)
-		usage := float64(bufferInfo.TotalMessages) / float64(bufferLength)
-		if x := (float64(bufferInfo.PendingCount) + float64(bufferInfo.AckPendingCount)) / float64(bufferLength); x < usage {
-			usage = x
-		}
-		b := &daemon.BufferInfo{
-			Pipeline:         &ps.pipeline.Name,
-			BufferName:       pointer.String(fmt.Sprintf("%v", buffer)),
-			PendingCount:     &bufferInfo.PendingCount,
-			AckPendingCount:  &bufferInfo.AckPendingCount,
-			TotalMessages:    &bufferInfo.TotalMessages,
-			BufferLength:     &bufferLength,
-			BufferUsageLimit: &bufferUsageLimit,
-			BufferUsage:      &usage,
-			IsFull:           pointer.Bool(usage >= bufferUsageLimit),
-		}
-		buffers = append(buffers, b)
+func (ps *PipelineMetadataQuery) ListBuffers(ctx context.Context, req *daemon.ListBuffersRequest) (*daemon.ListBuffersResponse, error) {
+	resp, err := listBuffers(ctx, ps.pipeline, ps.isbSvcClient)
+	if err != nil {
+		return nil, err
 	}
-	resp.Buffers = buffers
 	return resp, nil
 }
 
 // GetBuffer is used to obtain one buffer information of a pipeline
-func (ps *pipelineMetadataQuery) GetBuffer(ctx context.Context, req *daemon.GetBufferRequest) (*daemon.GetBufferResponse, error) {
+func (ps *PipelineMetadataQuery) GetBuffer(ctx context.Context, req *daemon.GetBufferRequest) (*daemon.GetBufferResponse, error) {
 	bufferInfo, err := ps.isbSvcClient.GetBufferInfo(ctx, *req.Buffer)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get information of buffer %q:%v", *req.Buffer, err)
@@ -157,7 +134,7 @@ func (ps *pipelineMetadataQuery) GetBuffer(ctx context.Context, req *daemon.GetB
 // Response contains the metrics for each partition of the vertex.
 // In the future maybe latency will also be added here?
 // Should this method live here or maybe another file?
-func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daemon.GetVertexMetricsRequest) (*daemon.GetVertexMetricsResponse, error) {
+func (ps *PipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daemon.GetVertexMetricsRequest) (*daemon.GetVertexMetricsResponse, error) {
 	resp := new(daemon.GetVertexMetricsResponse)
 
 	abstractVertex := ps.pipeline.GetVertex(req.GetVertex())
@@ -186,7 +163,7 @@ func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daem
 }
 
 // getPending returns the pending count for each partition of the vertex
-func (ps *pipelineMetadataQuery) getPending(ctx context.Context, req *daemon.GetVertexMetricsRequest) map[string]map[string]int64 {
+func (ps *PipelineMetadataQuery) getPending(ctx context.Context, req *daemon.GetVertexMetricsRequest) map[string]map[string]int64 {
 	vertexName := fmt.Sprintf("%s-%s", ps.pipeline.Name, req.GetVertex())
 	log := logging.FromContext(ctx)
 
@@ -247,57 +224,14 @@ func (ps *pipelineMetadataQuery) getPending(ctx context.Context, req *daemon.Get
 	return totalPendingMap
 }
 
-func (ps *pipelineMetadataQuery) GetPipelineStatus(ctx context.Context, req *daemon.GetPipelineStatusRequest) (*daemon.GetPipelineStatusResponse, error) {
-
+func (ps *PipelineMetadataQuery) GetPipelineStatus(ctx context.Context, req *daemon.GetPipelineStatusRequest) (*daemon.GetPipelineStatusResponse, error) {
+	status := ps.healthChecker.getCurrentHealth()
 	resp := new(daemon.GetPipelineStatusResponse)
-
-	// get all vertices of pipeline
-	vertices := ps.pipeline.Spec.Vertices
-
-	// loop over vertices and get metrics to check pending messages vs processing rate
-	for _, vertex := range vertices {
-		vertexReq := new(daemon.GetVertexMetricsRequest)
-		vertexReq.Vertex = &vertex.Name
-		vertexResp, err := ps.GetVertexMetrics(ctx, vertexReq)
-		// if err is not nil, more than likely autoscaling is down to 0 and metrics are not available
-		if err != nil {
-			resp.Status = &daemon.PipelineStatus{
-				Status:  pointer.String(PipelineStatusUnknown),
-				Message: pointer.String("Pipeline status is unknown."),
-			}
-			return resp, nil
-		}
-
-		totalProcessingRate := float64(0)
-		totalPending := int64(0)
-		// may need to revisit later, another concern could be that the processing rate is too slow instead of just 0
-		for _, vertexMetrics := range vertexResp.VertexMetrics {
-			if vertexMetrics.GetProcessingRates() != nil {
-				if p, ok := vertexMetrics.GetProcessingRates()["default"]; ok {
-					totalProcessingRate += p
-				}
-			}
-			if vertexMetrics.GetPendings() != nil {
-				if p, ok := vertexMetrics.GetPendings()["default"]; ok {
-					totalPending += p
-				}
-			}
-		}
-
-		if totalPending > 0 && totalProcessingRate == 0 {
-			resp.Status = &daemon.PipelineStatus{
-				Status:  pointer.String(PipelineStatusError),
-				Message: pointer.String(fmt.Sprintf("Pipeline has an error. Vertex %s is not processing pending messages.", vertex.Name)),
-			}
-			return resp, nil
-		}
-	}
-
 	resp.Status = &daemon.PipelineStatus{
-		Status:  pointer.String(PipelineStatusOK),
-		Message: pointer.String("Pipeline has no issue."),
+		Status:  pointer.String(status.Status),
+		Message: pointer.String(status.Message),
+		Code:    pointer.String(status.Code),
 	}
-
 	return resp, nil
 }
 
@@ -315,3 +249,47 @@ func getBufferLimits(pl *v1alpha1.Pipeline, v v1alpha1.AbstractVertex) (bufferLe
 	}
 	return bufferLength, bufferUsageLimit
 }
+
+// listBuffers returns the list of ISB buffers for the pipeline and their information
+// We use the isbSvcClient to get the buffer information
+func listBuffers(ctx context.Context, pipeline *v1alpha1.Pipeline, isbSvcClient isbsvc.ISBService) (*daemon.ListBuffersResponse, error) {
+	log := logging.FromContext(ctx)
+	resp := new(daemon.ListBuffersResponse)
+
+	buffers := []*daemon.BufferInfo{}
+	for _, buffer := range pipeline.GetAllBuffers() {
+		bufferInfo, err := isbSvcClient.GetBufferInfo(ctx, buffer)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get information of buffer %q", buffer)
+		}
+		log.Debugf("Buffer %s has bufferInfo %+v", buffer, bufferInfo)
+		v := pipeline.FindVertexWithBuffer(buffer)
+		if v == nil {
+			return nil, fmt.Errorf("unexpected error, buffer %q not found from the pipeline", buffer)
+		}
+		bufferLength, bufferUsageLimit := getBufferLimits(pipeline, *v)
+		usage := float64(bufferInfo.TotalMessages) / float64(bufferLength)
+		if x := (float64(bufferInfo.PendingCount) + float64(bufferInfo.AckPendingCount)) / float64(bufferLength); x < usage {
+			usage = x
+		}
+		b := &daemon.BufferInfo{
+			Pipeline:         &pipeline.Name,
+			BufferName:       pointer.String(fmt.Sprintf("%v", buffer)),
+			PendingCount:     &bufferInfo.PendingCount,
+			AckPendingCount:  &bufferInfo.AckPendingCount,
+			TotalMessages:    &bufferInfo.TotalMessages,
+			BufferLength:     &bufferLength,
+			BufferUsageLimit: &bufferUsageLimit,
+			BufferUsage:      &usage,
+			IsFull:           pointer.Bool(usage >= bufferUsageLimit),
+		}
+		buffers = append(buffers, b)
+	}
+	resp.Buffers = buffers
+	return resp, nil
+}
+
+// StartHealthCheck starts the health check for the pipeline using the health checker
+func (ps *PipelineMetadataQuery) StartHealthCheck(ctx context.Context) {
+	ps.healthChecker.startHealthCheck(ctx)
+}
diff --git a/pkg/daemon/server/service/pipeline_watermark_query.go b/pkg/daemon/server/service/pipeline_watermark_query.go
index c86ce8df..bb1ff4c2 100644
--- a/pkg/daemon/server/service/pipeline_watermark_query.go
+++ b/pkg/daemon/server/service/pipeline_watermark_query.go
@@ -71,7 +71,7 @@ func BuildWatermarkStores(ctx context.Context, pipeline *v1alpha1.Pipeline, isbs
 }
 
 // GetPipelineWatermarks is used to return the head watermarks for a given pipeline.
-func (ps *pipelineMetadataQuery) GetPipelineWatermarks(ctx context.Context, request *daemon.GetPipelineWatermarksRequest) (*daemon.GetPipelineWatermarksResponse, error) {
+func (ps *PipelineMetadataQuery) GetPipelineWatermarks(ctx context.Context, request *daemon.GetPipelineWatermarksRequest) (*daemon.GetPipelineWatermarksResponse, error) {
 	resp := new(daemon.GetPipelineWatermarksResponse)
 	isWatermarkEnabled := !ps.pipeline.Spec.Watermark.Disabled
 
diff --git a/pkg/shared/ewma/interface.go b/pkg/shared/ewma/interface.go
new file mode 100644
index 00000000..4ad31168
--- /dev/null
+++ b/pkg/shared/ewma/interface.go
@@ -0,0 +1,14 @@
+package ewma
+
+// EWMA is the interface for Exponentially Weighted Moving Average
+// It is used to calculate the moving average with decay of a series of numbers
+type EWMA interface {
+	// Add adds a new value to the EWMA
+	Add(float64)
+	// Get returns the current value of the EWMA
+	Get() float64
+	// Reset resets the EWMA to the initial value
+	Reset()
+	// Set sets the EWMA to the given value
+	Set(float64)
+}
diff --git a/pkg/shared/ewma/simple_ewma.go b/pkg/shared/ewma/simple_ewma.go
new file mode 100644
index 00000000..f7dcbd36
--- /dev/null
+++ b/pkg/shared/ewma/simple_ewma.go
@@ -0,0 +1,56 @@
+package ewma
+
+const (
+	defaultAlpha = 30.0
+	// constDecayFactor is the default decay factor
+	constDecayFactor = 2.0 / (defaultAlpha + 1.0)
+)
+
+// SimpleEWMA is a simple implementation of EWMA
+type SimpleEWMA struct {
+	// alpha is the smoothing factor
+	alpha float64
+	// value is the current value of the EWMA
+	value float64
+	// init is a flag to indicate if the EWMA has been initialized
+	init bool
+}
+
+// NewSimpleEWMA returns a new SimpleEWMA
+// If the alpha is not provided we use a default value of constDecayFactor
+// If the alpha is provided we calulate the smoothing factor from it
+func NewSimpleEWMA(alpha ...float64) *SimpleEWMA {
+	if len(alpha) > 0 {
+		decay := 2.0 / (alpha[0] + 1.0)
+		return &SimpleEWMA{alpha: decay}
+	}
+	return &SimpleEWMA{alpha: constDecayFactor}
+}
+
+// Add adds a new value to the EWMA
+func (s *SimpleEWMA) Add(value float64) {
+	// If the EWMA has not been initialized, set the value and return
+	if !s.init {
+		s.value = value
+		s.init = true
+		return
+	}
+	// Otherwise, calculate the EWMA
+	s.value = s.value + s.alpha*(value-s.value)
+}
+
+// Get returns the current value of the EWMA
+func (s *SimpleEWMA) Get() float64 {
+	return s.value
+}
+
+// Reset resets the EWMA to the initial value
+func (s *SimpleEWMA) Reset() {
+	s.value = 0
+	s.init = false
+}
+
+// Set sets the EWMA to the given value
+func (s *SimpleEWMA) Set(value float64) {
+	s.value = value
+}
diff --git a/pkg/shared/health-status-code/code_map.go b/pkg/shared/health-status-code/code_map.go
new file mode 100644
index 00000000..44af3956
--- /dev/null
+++ b/pkg/shared/health-status-code/code_map.go
@@ -0,0 +1,108 @@
+package health_status_code
+
+// HealthCodeInfo is used to maintain status codes for vertex level health
+type HealthCodeInfo struct {
+	Status      string
+	Criticality string
+}
+
+// newHealthCodeInfo is used to create a new HealthCodeInfo object
+func newHealthCodeInfo(status string, criticality string) *HealthCodeInfo {
+	return &HealthCodeInfo{
+		Status:      status,
+		Criticality: criticality,
+	}
+}
+
+// VertexHealthMap is used to maintain status codes for vertex level health
+// Each map entry is a map of status code as the key to the status message and the criticality of the status.
+// Status codes are in incremental like
+// 1. V1
+// 2. V2
+// 3. V3
+// The criticality is used to determine the overall status of the pipeline
+// Criticality can be one of the following:
+// 1. Critical: The pipeline is in a critical state
+// 2. Warning: The pipeline is in a warning state
+// 3. Healthy: The pipeline is healthy
+// 4. Unknown: The pipeline is in an unknown state
+var vertexHealthMap = map[string]*HealthCodeInfo{
+	"V1": newHealthCodeInfo(
+		"All pods are running",
+		"Healthy",
+	),
+	"V2": newHealthCodeInfo(
+		"Vertex is not in running state",
+		"Critical",
+	),
+	"V3": newHealthCodeInfo(
+		"Vertex is in running but containers are not in running state",
+		"Warning",
+	),
+	"V4": newHealthCodeInfo(
+		"All vertices healthy in the pipeline",
+		"Healthy",
+	),
+	"V5": newHealthCodeInfo(
+		"One or more vertices are unhealthy in the pipeline",
+		"Warning",
+	),
+	"V6": newHealthCodeInfo(
+		"Pipeline is in an unknown state",
+		"Critical",
+	),
+	"V7": newHealthCodeInfo(
+		"Pipeline is paused",
+		"Healthy",
+	),
+	"V8": newHealthCodeInfo(
+		"Pipeline is killed",
+		"Healthy",
+	),
+	"V9": newHealthCodeInfo(
+		"Vertex not able to get the desired number of replicas",
+		"Warning",
+	),
+}
+
+// DataflowHealthCodeMap is used to maintain status codes for dataflow level health
+// Each map entry is a map of status code as the key to the status message and the criticality of the status.
+// Status codes are in incremental like
+// 1. D1
+// 2. D2
+// 3. D3
+// The criticality is used to determine the overall status of the pipeline
+// Criticality can be one of the following:
+// 1. Critical: The pipeline is in a critical state
+// 2. Warning: The pipeline is in a warning state
+// 3. Healthy: The pipeline is healthy
+
+var dataflowHealthMap = map[string]*HealthCodeInfo{
+	"D1": newHealthCodeInfo(
+		"Dataflow is healthy",
+		"Healthy",
+	),
+	"D2": newHealthCodeInfo(
+		"Dataflow in warning state for one or more vertices",
+		"Warning",
+	),
+	"D3": newHealthCodeInfo(
+		"Dataflow in critical state for one or more vertices",
+		"Critical",
+	),
+	"D4": newHealthCodeInfo(
+		"Dataflow in unknown state",
+		"Critical",
+	),
+}
+
+// getHealthCodeInfo is used to get the status code information for a given status code
+func getHealthCodeInfo(code string) *HealthCodeInfo {
+	if status, ok := vertexHealthMap[code]; ok {
+		return status
+	}
+	if status, ok := dataflowHealthMap[code]; ok {
+		return status
+	}
+	return nil
+}
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index 119595d9..a726468c 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -65,6 +65,7 @@ type handler struct {
 	daemonClientsCache   *lru.Cache[string, *daemonclient.DaemonClient]
 	dexObj               *DexObject
 	localUsersAuthObject *LocalUsersAuthObject
+	healthChecker        *HealthChecker
 }
 
 // NewHandler is used to provide a new instance of the handler type
@@ -93,6 +94,7 @@ func NewHandler(dexObj *DexObject, localUsersAuthObject *LocalUsersAuthObject) (
 		daemonClientsCache:   daemonClientsCache,
 		dexObj:               dexObj,
 		localUsersAuthObject: localUsersAuthObject,
+		healthChecker:        NewHealthChecker(),
 	}, nil
 }
 
@@ -876,17 +878,47 @@ func (h *handler) GetNamespaceEvents(c *gin.Context) {
 	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, response))
 }
 
-// GetPipelineStatus returns the pipeline status. It is based on Health and Criticality.
-// Health can be "healthy (0) | unhealthy (1) | paused (3) | unknown (4)".
-// Health here indicates pipeline's ability to process messages.
-// A backlogged pipeline can be healthy even though it has an increasing back-pressure. Health purely means it is up and running.
-// Pipelines health will be the max(health) based of each vertex's health
-// Criticality on the other end shows whether the pipeline is working as expected.
+// GetPipelineStatus returns the pipeline status. It is based on Resource Health and Data Criticality.
+// Resource Health can be "healthy (0) | unhealthy (1) | paused (3) | unknown (4)".
+// A backlogged pipeline can be healthy even though it has an increasing back-pressure.
+// Resource Health purely means it is up and running.
+// Resource health will be the max(health) based of each vertex's health
+// Data Criticality on the other end shows whether the pipeline is working as expected.
 // It represents the pending messages, lags, etc.
-// Criticality can be "ok (0) | warning (1) | critical (2)".
-// Health and Criticality are different because ...?
+// Data Criticality can be "ok (0) | warning (1) | critical (2)".
+// GetPipelineStatus is used to return the status of a given pipeline
+// It is divided into two parts:
+// 1. Pipeline Resource Health: It is based on the health of each vertex in the pipeline
+// 2. Data Criticality: It is based on the data movement of the pipeline
 func (h *handler) GetPipelineStatus(c *gin.Context) {
-	c.JSON(http.StatusNotImplemented, "working on it")
+	ns, pipeline := c.Param("namespace"), c.Param("pipeline")
+
+	// Get the vertex level health of the pipeline
+	resourceHealth, err := h.healthChecker.getPipelineResourceHealth(h, ns, pipeline)
+	if err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to get the dataStatus for pipeline %q: %s", pipeline, err.Error()))
+		return
+	}
+
+	// Get a new daemon client for the given pipeline
+	client, err := h.getDaemonClient(ns, pipeline)
+	if err != nil || client == nil {
+		h.respondWithError(c, fmt.Sprintf("failed to get daemon service client for pipeline %q, %s", pipeline, err.Error()))
+		return
+	}
+	// Get the data criticality for the given pipeline
+	dataStatus, err := client.GetPipelineStatus(context.Background(), pipeline)
+	if err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to get the dataStatus for pipeline %q: %s", pipeline, err.Error()))
+		return
+	}
+
+	// Create a response string based on the vertex health and data criticality
+	// We combine both the states to get the final dataStatus of the pipeline
+	response := NewHealthResponse(resourceHealth.Status, dataStatus.GetStatus(),
+		resourceHealth.Message, dataStatus.GetMessage(), resourceHealth.Code, dataStatus.GetCode())
+
+	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, response))
 }
 
 // getAllNamespaces is a utility used to fetch all the namespaces in the cluster
diff --git a/server/apis/v1/health.go b/server/apis/v1/health.go
new file mode 100644
index 00000000..f52c45f3
--- /dev/null
+++ b/server/apis/v1/health.go
@@ -0,0 +1,209 @@
+package v1
+
+import (
+	"fmt"
+	"time"
+
+	evictCache "github.com/hashicorp/golang-lru/v2/expirable"
+	"golang.org/x/net/context"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+)
+
+const (
+	// resourceCacheRefreshDuration is the duration after which the vertex status cache is refreshed
+	resourceCacheRefreshDuration = 30 * time.Second
+)
+
+// resourceHealthResponse is the response returned by the vertex health check API
+type resourceHealthResponse struct {
+	// Status is the overall vertex status of the pipeline
+	Status string `json:"status"`
+	// Message is the error message if any
+	Message string `json:"message"`
+	// Code is the status code for the vertex health
+	Code string `json:"code"`
+}
+
+// HealthChecker is the struct to hold the resource status cache for the pipeline
+type HealthChecker struct {
+	resourceStatusCache *evictCache.LRU[string, *resourceHealthResponse]
+}
+
+// NewHealthChecker is used to create a new health checker
+func NewHealthChecker() *HealthChecker {
+	c := evictCache.NewLRU[string, *resourceHealthResponse](500, nil, resourceCacheRefreshDuration)
+	return &HealthChecker{
+		resourceStatusCache: c,
+	}
+}
+
+// getPipelineResourceHealth is used to provide the overall vertex health and status of the pipeline
+// This first check if the pipeline status is cached, if not, it checks for the current pipeline status
+// and caches it.
+func (hc *HealthChecker) getPipelineResourceHealth(h *handler, ns string,
+	pipeline string) (*resourceHealthResponse, error) {
+	ctx := context.Background()
+	log := logging.FromContext(ctx)
+
+	// create a cache key for the pipeline
+	// It is a combination of namespace and pipeline name
+	// In the form of <namespace>-<pipeline>
+	cacheKey := fmt.Sprintf("%s-%s", ns, pipeline)
+
+	// check if the pipeline status is cached
+	if status, ok := hc.resourceStatusCache.Get(cacheKey); ok {
+		log.Info("Pipeline status from cache: ", status)
+		return status, nil
+	}
+	// if not present in cache, check for the current pipeline status
+	status, err := checkVertexLevelHealth(ctx, h, ns, pipeline)
+	if err != nil {
+		return status, err
+	}
+	// update cache with the new pipeline status
+	hc.resourceStatusCache.Add(cacheKey, status)
+
+	return status, nil
+}
+
+// checkVertexLevelHealth is used to provide the overall vertex health and status of the pipeline
+// They can be of the following types:
+// 1. Healthy: All the vertices are healthy
+// 2. Unhealthy: One or more vertices are unhealthy
+// 3. Paused: The pipeline is paused
+// 4. Unknown: The pipeline is in an unknown state
+// 5. Killed: The pipeline is killed
+// To check for vertex level status we need to check for two things,
+// 1) the number of replicas running in the vertex
+// are equal to the number of desired replicas and the pods are in running state
+// 2) If all the containers in the pod are in running state
+// if any of the above conditions are not met, the vertex is unhealthy
+func checkVertexLevelHealth(ctx context.Context, h *handler, ns string,
+	pipeline string) (*resourceHealthResponse, error) {
+	log := logging.FromContext(ctx)
+	// get the pipeline object
+	pl, err := h.numaflowClient.Pipelines(ns).Get(context.Background(), pipeline, metav1.GetOptions{})
+	// if error return unknown status
+	if err != nil {
+		return &resourceHealthResponse{
+			Status:  PipelineStatusUnknown,
+			Message: fmt.Sprintf("error in getting pipeline %q status: %v", pipeline, err),
+			Code:    "V6",
+		}, err
+	}
+
+	// if the pipeline is paused, return inactive status
+	// this cannot be checked at individual vertex level, hence needs to be checked here
+	if pl.Spec.Lifecycle.GetDesiredPhase() == dfv1.PipelinePhasePaused {
+		return &resourceHealthResponse{
+			Status:  PipelineStatusInactive,
+			Message: fmt.Sprintf("pipeline %q is paused", pipeline),
+			Code:    "V7",
+		}, nil
+	}
+
+	// if the pipeline is killed, return killed status
+	// this cannot be checked at individual vertex level, hence needs to be checked here
+	if pl.Spec.Lifecycle.GetDesiredPhase() == dfv1.PipelinePhaseDeleting {
+		return &resourceHealthResponse{
+			Status:  PipelineStatusDeleting,
+			Message: fmt.Sprintf("pipeline %q is killed", pipeline),
+			Code:    "V8",
+		}, nil
+	}
+
+	// get the list of vertices in the pipeline
+	vertices := pl.Spec.Vertices
+
+	// Iterate over all the vertices
+	for _, vertex := range vertices {
+		vertexName := fmt.Sprintf("%s-%s", pipeline, vertex.Name)
+		// fetch the current spec of the vertex
+		v, err := h.numaflowClient.Vertices(ns).Get(context.Background(), vertexName, metav1.GetOptions{})
+		if err != nil {
+			return &resourceHealthResponse{
+				Status:  PipelineStatusUnknown,
+				Message: fmt.Sprintf("error in getting vertex %q status: %v", vertexName, err),
+				Code:    "V6",
+			}, err
+		}
+		ok, resp, err := isVertexHealthy(h, ns, pipeline, v, vertex.Name)
+		if err != nil {
+			resp.Status = PipelineStatusUnknown
+			return resp, err
+		}
+		if !ok {
+			log.Infof("vertex %q is unhealthy: %s", vertex.Name, resp)
+			resp.Status = PipelineStatusUnhealthy
+			return resp, nil
+		}
+	}
+	// if all the vertices are healthy, return healthy status
+	return &resourceHealthResponse{
+		Status:  PipelineStatusHealthy,
+		Message: fmt.Sprintf("pipeline %q is healthy", pipeline),
+		Code:    "V4",
+	}, nil
+}
+
+// isVertexHealthy is used to check if the vertex is healthy or not
+// It checks for the following:
+// 1) If the vertex is in running state
+// 2) the number of replicas running in the vertex
+// are equal to the number of desired replicas and the pods are in running state
+// 3) If all the containers in the pod are in running state
+// if any of the above conditions are not met, the vertex is unhealthy
+// Based on the above conditions, it returns the status code and message
+func isVertexHealthy(h *handler, ns string, pipeline string, vertex *dfv1.Vertex,
+	vertexName string) (bool, *resourceHealthResponse, error) {
+	// check if the vertex is in running state
+	if vertex.Status.Phase != dfv1.VertexPhaseRunning {
+		// check if the number of replicas running in the vertex
+		// are equal to the number of desired replicas
+		if int(vertex.Status.Replicas) != vertex.GetReplicas() {
+			return false, &resourceHealthResponse{
+				Message: fmt.Sprintf("vertex %q has %d replicas running, "+
+					"expected %d", vertex.Name, vertex.Status.Replicas, vertex.GetReplicas()),
+				Code: "V9",
+			}, nil
+		}
+		// Else return the error message from the status
+		return false, &resourceHealthResponse{
+			Message: fmt.Sprintf("error in vertex %s", vertex.Status.Message),
+			Code:    "V2",
+		}, nil
+	}
+
+	// Get all the pods for the given vertex
+	pods, err := h.kubeClient.CoreV1().Pods(ns).List(context.Background(), metav1.ListOptions{
+		LabelSelector: fmt.Sprintf("%s=%s,%s=%s", dfv1.KeyPipelineName,
+			pipeline, dfv1.KeyVertexName, vertexName),
+	})
+	if err != nil {
+		return false, &resourceHealthResponse{
+			Message: fmt.Sprintf("error in getting pods for vertex %q: %v", vertexName, err),
+			Code:    "V6",
+		}, err
+	}
+	// Iterate over all the pods, and verify if all the containers in the pod are in running state
+	for _, pod := range pods.Items {
+		// Iterate over all the containers in the pod
+		for _, containerStatus := range pod.Status.ContainerStatuses {
+			// if the container is not in running state, return false
+			if containerStatus.State.Running == nil {
+				return false, &resourceHealthResponse{
+					Message: fmt.Sprintf("container %q in pod %q is not running",
+						containerStatus.Name, pod.Name),
+					Code: "V3",
+				}, nil
+			}
+		}
+	}
+	return true, &resourceHealthResponse{
+		Message: fmt.Sprintf("vertex %q is healthy", vertexName),
+		Code:    "V1",
+	}, nil
+}
diff --git a/server/apis/v1/response_health.go b/server/apis/v1/response_health.go
new file mode 100644
index 00000000..a8db4c12
--- /dev/null
+++ b/server/apis/v1/response_health.go
@@ -0,0 +1,43 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+// HealthResponse is the response payload for health API.
+// It contains the health status of the vertex and data.
+// We include the Status, Message and Code for both resources and data.
+type HealthResponse struct {
+	ResourceHealthStatus  string `json:"resourceHealthStatus"`
+	DataHealthStatus      string `json:"dataHealthStatus"`
+	ResourceHealthMessage string `json:"resourceHealthMessage"`
+	DataHealthMessage     string `json:"dataHealthMessage"`
+	ResourceHealthCode    string `json:"resourceHealthCode"`
+	DataHealthCode        string `json:"dataHealthCode"`
+}
+
+// NewHealthResponse returns a HealthResponse object for the given status, message and code.
+func NewHealthResponse(resourceHealthStatus string, dataHealthStatus string,
+	resourceHealthMessage string, dataHealthMessage string, resourceHealthCode string,
+	dataHealthCode string) HealthResponse {
+	return HealthResponse{
+		ResourceHealthStatus:  resourceHealthStatus,
+		DataHealthStatus:      dataHealthStatus,
+		ResourceHealthMessage: resourceHealthMessage,
+		DataHealthMessage:     dataHealthMessage,
+		ResourceHealthCode:    resourceHealthCode,
+		DataHealthCode:        dataHealthCode,
+	}
+}
diff --git a/server/apis/v1/response_pipeline.go b/server/apis/v1/response_pipeline.go
index 1f0aaba8..9e1b0500 100644
--- a/server/apis/v1/response_pipeline.go
+++ b/server/apis/v1/response_pipeline.go
@@ -19,10 +19,13 @@ package v1
 import "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 
 const (
-	PipelineStatusHealthy  = "healthy"
-	PipelineStatusCritical = "critical"
-	PipelineStatusWarning  = "warning"
-	PipelineStatusInactive = "inactive"
+	PipelineStatusHealthy   = "healthy"
+	PipelineStatusCritical  = "critical"
+	PipelineStatusWarning   = "warning"
+	PipelineStatusInactive  = "inactive"
+	PipelineStatusUnknown   = "unknown"
+	PipelineStatusDeleting  = "deleting"
+	PipelineStatusUnhealthy = "unhealthy"
 )
 
 // Pipelines is a list of pipelines


Test patch:
diff --git a/pkg/daemon/server/service/pipeline_metrics_query_test.go b/pkg/daemon/server/service/pipeline_metrics_query_test.go
index 8fcbbce3..56aefb32 100644
--- a/pkg/daemon/server/service/pipeline_metrics_query_test.go
+++ b/pkg/daemon/server/service/pipeline_metrics_query_test.go
@@ -25,7 +25,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/utils/pointer"
 
 	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/apis/proto/daemon"
@@ -215,87 +214,3 @@ func TestListBuffers(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, len(resp.Buffers), 2)
 }
-
-// mock rater
-type mockRater_TestGetPipelineStatus struct {
-	isActivelyProcessing bool
-}
-
-func (mr *mockRater_TestGetPipelineStatus) Start(ctx context.Context) error {
-	return nil
-}
-
-func (mr *mockRater_TestGetPipelineStatus) GetRates(vertexName string, partitionName string) map[string]float64 {
-	res := make(map[string]float64)
-	if mr.isActivelyProcessing {
-		res["default"] = 4.894736842105263
-		res["1m"] = 5.084745762711864
-		res["5m"] = 4.894736842105263
-		res["15m"] = 4.894736842105263
-	} else {
-		res["default"] = 0
-		res["1m"] = 0
-		res["5m"] = 0
-		res["15m"] = 0
-	}
-	return res
-}
-
-func TestGetPipelineStatus(t *testing.T) {
-	pipelineName := "simple-pipeline"
-	pipeline := &v1alpha1.Pipeline{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: pipelineName,
-		},
-		Spec: v1alpha1.PipelineSpec{
-			Vertices: []v1alpha1.AbstractVertex{
-				{Name: "cat"},
-			},
-		},
-	}
-	client, _ := isbsvc.NewISBJetStreamSvc(pipelineName)
-	metricsResponse := `# HELP vertex_pending_messages Average pending messages in the last period of seconds. It is the pending messages of a vertex, not a pod.
-# TYPE vertex_pending_messages gauge
-vertex_pending_messages{period="15m",partition_name="-simple-pipeline-cat-0",pipeline="simple-pipeline",vertex="cat"} 4.011
-vertex_pending_messages{period="1m",partition_name="-simple-pipeline-cat-0",pipeline="simple-pipeline",vertex="cat"} 5.333
-vertex_pending_messages{period="5m",partition_name="-simple-pipeline-cat-0",pipeline="simple-pipeline",vertex="cat"} 6.002
-vertex_pending_messages{period="default",partition_name="-simple-pipeline-cat-0",pipeline="simple-pipeline",vertex="cat"} 7.00002
-`
-	req := &daemon.GetPipelineStatusRequest{Pipeline: &pipelineName}
-
-	// test when rater is actively processing
-	activeRater := &mockRater_TestGetPipelineStatus{isActivelyProcessing: true}
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil, activeRater)
-	assert.NoError(t, err)
-	ioReader := io.NopCloser(bytes.NewReader([]byte(metricsResponse)))
-	pipelineMetricsQueryService.httpClient = &mockHttpClient{
-		MockGet: func(url string) (*http.Response, error) {
-			return &http.Response{
-				StatusCode: 200,
-				Body:       ioReader,
-			}, nil
-		},
-	}
-	resp, err := pipelineMetricsQueryService.GetPipelineStatus(context.Background(), req)
-	assert.NoError(t, err)
-	OKPipelineResponse := daemon.PipelineStatus{Status: pointer.String("OK"), Message: pointer.String("Pipeline has no issue.")}
-	assert.Equal(t, &OKPipelineResponse, resp.Status)
-
-	// test when rater is not actively processing
-	idleRater := &mockRater_TestGetPipelineStatus{isActivelyProcessing: false}
-	pipelineMetricsQueryService, err = NewPipelineMetadataQuery(client, pipeline, nil, idleRater)
-	assert.NoError(t, err)
-	ioReader = io.NopCloser(bytes.NewReader([]byte(metricsResponse)))
-	pipelineMetricsQueryService.httpClient = &mockHttpClient{
-		MockGet: func(url string) (*http.Response, error) {
-			return &http.Response{
-				StatusCode: 200,
-				Body:       ioReader,
-			}, nil
-		},
-	}
-	resp, err = pipelineMetricsQueryService.GetPipelineStatus(context.Background(), req)
-	assert.NoError(t, err)
-	ErrorPipelineResponse := daemon.PipelineStatus{Status: pointer.String("Error"), Message: pointer.String("Pipeline has an error. Vertex cat is not processing pending messages.")}
-	assert.Equal(t, &ErrorPipelineResponse, resp.Status)
-}
diff --git a/pkg/shared/ewma/simple_ewma_test.go b/pkg/shared/ewma/simple_ewma_test.go
new file mode 100644
index 00000000..392cda46
--- /dev/null
+++ b/pkg/shared/ewma/simple_ewma_test.go
@@ -0,0 +1,39 @@
+package ewma
+
+import (
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	defaultEWMA = 77.14075212282631
+	span15EWMA  = 74.9112723022807
+)
+
+var samples = [14]float64{
+	83.92333333333333, 0, 83.24000000000001, 88.24, 77.61, 76.57333333333334, 79.91333333333334, 80.34,
+	74.90666666666667, 69.90666666666667, 71.65, 73.19333333333333, 72.18666666666667, 74.90666666666667,
+}
+
+// TestSimpleEWMA tests the SimpleEWMA implementation.
+func TestSimpleEWMA(t *testing.T) {
+	// Create a new EWMA with the default decay factor.
+	newEwma := NewSimpleEWMA()
+	for _, f := range samples {
+		newEwma.Add(f)
+	}
+	// Check if the value is within the margin of error.
+	assert.True(t, math.Abs(defaultEWMA-newEwma.Get()) < 0.00000001)
+
+	// Create a new EWMA with a custom decay factor.
+	newEwma = NewSimpleEWMA(15)
+	for _, f := range samples {
+		newEwma.Add(f)
+	}
+	// Check if the value is within the margin of error.
+	assert.True(t, math.Abs(span15EWMA-newEwma.Get()) < 0.00000001)
+}
+
+// TestSimpleEWMAInit tests the SimpleEWMA initialization.
diff --git a/pkg/shared/health-status-code/code_map_test.go b/pkg/shared/health-status-code/code_map_test.go
new file mode 100644
index 00000000..6444615b
--- /dev/null
+++ b/pkg/shared/health-status-code/code_map_test.go
@@ -0,0 +1,92 @@
+package health_status_code
+
+import "testing"
+
+func TestGetHealthCodeInfo(t *testing.T) {
+	type args struct {
+		code string
+	}
+	tests := []struct {
+		name string
+		args args
+		want *HealthCodeInfo
+	}{
+		{
+			name: "TestGetHealthCodeInfo",
+			args: args{
+				code: "V1",
+			},
+			want: &HealthCodeInfo{
+				Status:      "All pods are running",
+				Criticality: "Healthy",
+			},
+		},
+		{
+			name: "TestGetHealthCodeInfo",
+			args: args{
+				code: "V2",
+			},
+			want: &HealthCodeInfo{
+				Status:      "Vertex is not in running state",
+				Criticality: "Critical",
+			},
+		},
+		{
+			name: "TestGetHealthCodeInfo",
+			args: args{
+				code: "V3",
+			},
+			want: &HealthCodeInfo{
+				Status:      "Vertex is in running but containers are not in running state",
+				Criticality: "Warning",
+			},
+		},
+		{
+			name: "TestGetHealthCodeInfo",
+			args: args{
+				code: "V4",
+			},
+			want: &HealthCodeInfo{
+				Status:      "All vertices healthy in the pipeline",
+				Criticality: "Healthy",
+			},
+		},
+		{
+			name: "TestGetHealthCodeInfo",
+			args: args{
+				code: "V5",
+			},
+			want: &HealthCodeInfo{
+				Status:      "One or more vertices are unhealthy in the pipeline",
+				Criticality: "Warning",
+			},
+		},
+		{
+			name: "TestGetHealthCodeInfo",
+			args: args{
+				code: "V6",
+			},
+			want: &HealthCodeInfo{
+				Status:      "Pipeline is in an unknown state",
+				Criticality: "Critical",
+			},
+		},
+		{
+			name: "TestGetHealthCodeInfo",
+			args: args{
+				code: "V7",
+			},
+			want: &HealthCodeInfo{
+				Status:      "Pipeline is paused",
+				Criticality: "Healthy",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := getHealthCodeInfo(tt.args.code); got.Status != tt.want.Status {
+				t.Errorf("GetHealthCodeInfo() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/server/apis/v1/health_test.go b/server/apis/v1/health_test.go
new file mode 100644
index 00000000..4aa5f7bf
--- /dev/null
+++ b/server/apis/v1/health_test.go
@@ -0,0 +1,168 @@
+package v1
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/net/context"
+	appv1 "k8s.io/api/apps/v1"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/pointer"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+)
+
+const (
+	RunningPod = "running"
+	WaitingPod = "waiting"
+)
+
+// Client is the struct to hold the Kubernetes Clientset
+type Client struct {
+	Clientset kubernetes.Interface
+}
+
+func init() {
+	_ = dfv1.AddToScheme(scheme.Scheme)
+	_ = appv1.AddToScheme(scheme.Scheme)
+	//_ = corev1.AddToScheme(scheme.Scheme)
+}
+
+func (c Client) CreatePod(pod *v1.Pod) (*v1.Pod, error) {
+	_, err := c.Clientset.CoreV1().Pods(pod.Namespace).Create(context.TODO(), pod, metav1.CreateOptions{})
+	if err != nil {
+		fmt.Printf("Error occured while creating pod %s: %s", pod.Name, err.Error())
+		return nil, err
+	}
+
+	fmt.Printf("Pod %s is succesfully created", pod.Name)
+	return pod, nil
+}
+
+func createPod(phase string) {
+	client := Client{
+		Clientset: fakeKubeClient,
+	}
+	pod := fakePod("test-pl", "test-vertex", testNamespace, phase)
+	_, err := client.CreatePod(pod)
+	if err != nil {
+		fmt.Print(err.Error())
+	}
+	pod, err = client.Clientset.CoreV1().Pods(pod.Namespace).Get(context.Background(), pod.Name, metav1.GetOptions{})
+	if err != nil {
+		fmt.Printf("Error occured while getting pod %s: %s", pod.Name, err.Error())
+	}
+}
+
+func removePod() {
+	client := Client{
+		Clientset: fakeKubeClient,
+	}
+	pod := fakePod("test-pl", "test-vertex", testNamespace, "running")
+	err := client.Clientset.CoreV1().Pods(pod.Namespace).Delete(context.Background(), pod.Name, metav1.DeleteOptions{})
+	if err != nil {
+		fmt.Printf("Error occured while deleting pod %s: %s", pod.Name, err.Error())
+	}
+}
+
+// fakePipeline returns a fake pipeline for testing
+
+// TestIsVertexHealthy verifies the functionality of the vertex level health check
+// It creates a fake pipeline and checks the health of each vertex
+// We test multiple scenarios:
+// 1. All vertices are healthy
+// 2. One or more vertices are unhealthy
+// 3. The number of replicas running in the vertex is less than the expected number of replicas
+// 4. Vertex is not in running state
+func TestIsVertexHealthy(t *testing.T) {
+	// Test vertex is in running phase and the pods are in running state
+	t.Run("test all pods are healthy", func(t *testing.T) {
+		pipeline := fakePipeline()
+		vertexName := "test-vertex"
+		vertex := fakeVertex(vertexName, dfv1.VertexPhaseRunning)
+
+		// Create fake handler
+		h := &handler{
+			kubeClient:     fakeKubeClient,
+			numaflowClient: &fakeNumaClient,
+		}
+
+		// Create fake pod in running state
+		createPod(RunningPod)
+		defer removePod()
+		healthy, _, err := isVertexHealthy(h, testNamespace, pipeline.GetName(), vertex, vertexName)
+		if err != nil {
+			return
+		}
+		assert.True(t, healthy)
+	})
+
+	// Test vertex is in running phase and the pods are in waiting state
+	t.Run("test pod not running", func(t *testing.T) {
+		pipeline := fakePipeline()
+		vertexName := "test-vertex"
+		vertex := fakeVertex(vertexName, dfv1.VertexPhaseRunning)
+
+		// Create fake handler
+		h := &handler{
+			kubeClient:     fakeKubeClient,
+			numaflowClient: &fakeNumaClient,
+		}
+		createPod(WaitingPod)
+		defer removePod()
+		healthy, r, err := isVertexHealthy(h, testNamespace, pipeline.GetName(), vertex, vertexName)
+		if err != nil {
+			return
+		}
+		assert.False(t, healthy)
+		assert.Equal(t, "V3", r.Code)
+	})
+
+	// Test vertex is not in running state
+	t.Run("test vertex not in running phase", func(t *testing.T) {
+		pipeline := fakePipeline()
+		vertexName := "test-vertex"
+		// Create a fake vertex in failed state
+		vertex := fakeVertex(vertexName, dfv1.VertexPhaseFailed)
+
+		// Create fake handler
+		h := &handler{
+			kubeClient:     fakeKubeClient,
+			numaflowClient: &fakeNumaClient,
+		}
+		healthy, r, err := isVertexHealthy(h, testNamespace, pipeline.GetName(), vertex, vertexName)
+		if err != nil {
+			return
+		}
+		assert.False(t, healthy)
+		// Refer: pkg/shared/health-status-code
+		assert.Equal(t, "V2", r.Code)
+	})
+
+	// Test vertex replica count is not equal to the desired replica count
+	t.Run("test vertex replica count not equal to desired replica count", func(t *testing.T) {
+		pipeline := fakePipeline()
+		vertexName := "test-vertex"
+		// Create a fake vertex in failed state
+		vertex := fakeVertex(vertexName, dfv1.VertexPhaseFailed)
+		// Update the replica count to 2
+		vertex.Spec.Replicas = pointer.Int32(2)
+
+		// Create fake handler
+		h := &handler{
+			kubeClient:     fakeKubeClient,
+			numaflowClient: &fakeNumaClient,
+		}
+		healthy, r, err := isVertexHealthy(h, testNamespace, pipeline.GetName(), vertex, vertexName)
+		if err != nil {
+			return
+		}
+		assert.False(t, healthy)
+		// Refer: pkg/shared/health-status-code
+		assert.Equal(t, "V9", r.Code)
+	})
+}
diff --git a/server/apis/v1/test_utils.go b/server/apis/v1/test_utils.go
new file mode 100644
index 00000000..e4e0a8d4
--- /dev/null
+++ b/server/apis/v1/test_utils.go
@@ -0,0 +1,152 @@
+package v1
+
+import (
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	fakeclientset "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/utils/pointer"
+
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/client/clientset/versioned/typed/numaflow/v1alpha1/fake"
+)
+
+const testNamespace = "test-ns"
+
+var (
+	fakeNumaClient    = fake.FakeNumaflowV1alpha1{}
+	fakeKubeClient    = fakeclientset.NewSimpleClientset()
+	testContainerName = "test-container"
+)
+
+// getContainerStatus returns a container status with the given phase.
+func getContainerStatus(phase string) corev1.ContainerStatus {
+	switch phase {
+	case "running":
+		return corev1.ContainerStatus{
+			Name:  testContainerName,
+			Ready: true,
+			State: corev1.ContainerState{
+				Running: &corev1.ContainerStateRunning{
+					StartedAt: metav1.Time{Time: time.Now()},
+				},
+			},
+		}
+	case "waiting":
+		return corev1.ContainerStatus{
+			Name:  testContainerName,
+			Ready: false,
+			State: corev1.ContainerState{
+				Waiting: &corev1.ContainerStateWaiting{
+					Reason:  "test-reason",
+					Message: "test-message",
+				},
+			},
+		}
+	}
+	return corev1.ContainerStatus{}
+}
+
+// fakePod returns a fake pod with the given pipeline name, vertex name, namespace and phase.
+func fakePod(pipelineName string, vertexName string, namespace string, phase string) *corev1.Pod {
+	containerStatus := getContainerStatus(phase)
+	pod := &corev1.Pod{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Pod",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-pod",
+			Namespace: namespace,
+			Labels: map[string]string{
+				dfv1.KeyPipelineName: pipelineName,
+				dfv1.KeyVertexName:   vertexName,
+			},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name: testContainerName,
+				},
+			},
+		},
+		Status: corev1.PodStatus{
+			Phase:             corev1.PodPhase("Running"),
+			ContainerStatuses: []corev1.ContainerStatus{containerStatus},
+		},
+	}
+	return pod
+}
+
+func fakePipeline() *dfv1.Pipeline {
+	return &dfv1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-pl",
+			Namespace: testNamespace,
+		},
+		Spec: dfv1.PipelineSpec{
+			Vertices: []dfv1.AbstractVertex{
+				{
+					Name: "input",
+					Source: &dfv1.Source{
+						UDTransformer: &dfv1.UDTransformer{
+							Builtin: &dfv1.Transformer{Name: "filter"},
+						}},
+				},
+				{
+					Name: "map",
+					UDF: &dfv1.UDF{
+						Builtin: &dfv1.Function{Name: "cat"},
+					},
+				},
+				{
+					Name: "reduce",
+					UDF: &dfv1.UDF{
+						Container: &dfv1.Container{
+							Image: "test-image",
+						},
+						GroupBy: &dfv1.GroupBy{
+							Window: dfv1.Window{
+								Fixed: &dfv1.FixedWindow{Length: &metav1.Duration{
+									Duration: 60 * time.Second,
+								}},
+							},
+							Keyed: true,
+							Storage: &dfv1.PBQStorage{
+								EmptyDir: &corev1.EmptyDirVolumeSource{},
+							},
+						},
+					},
+				},
+				{
+					Name: "output",
+					Sink: &dfv1.Sink{},
+				},
+			},
+			Edges: []dfv1.Edge{
+				{From: "input", To: "map"},
+				{From: "map", To: "reduce"},
+				{From: "reduce", To: "output"},
+			},
+		},
+	}
+}
+
+// fakeVertex returns a fake vertex with the given name and phase.
+func fakeVertex(name string, phase dfv1.VertexPhase) *dfv1.Vertex {
+	v := &dfv1.Vertex{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: testNamespace,
+		},
+		Status: dfv1.VertexStatus{
+			Phase:    phase,
+			Replicas: 1,
+		},
+		Spec: dfv1.VertexSpec{
+			Replicas: pointer.Int32(1),
+		},
+	}
+	return v
+}