Repository: numaproj/numaflow
Pull request #114 (2022-07-14 17:31:23)

chore: Surface rate metrics in Daemon Svc
This PR exposes an API Server/ Daemon service endpoint to get the vertex processing rate which in turn queries the metrics service. 



Patch:
diff --git a/go.mod b/go.mod
index d6f547ee..e3fedd7e 100644
--- a/go.mod
+++ b/go.mod
@@ -25,6 +25,7 @@ require (
 	github.com/nats-io/nats.go v1.16.0
 	github.com/numaproj/numaflow-go v0.0.1
 	github.com/prometheus/client_golang v1.12.1
+	github.com/prometheus/common v0.32.1
 	github.com/soheilhy/cmux v0.1.5
 	github.com/spf13/cobra v1.2.1
 	github.com/spf13/viper v1.9.0
@@ -135,7 +136,6 @@ require (
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/rogpeppe/go-internal v1.8.0 // indirect
diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
index 6ddd5134..54f09710 100644
--- a/pkg/apis/numaflow/v1alpha1/const.go
+++ b/pkg/apis/numaflow/v1alpha1/const.go
@@ -101,6 +101,13 @@ const (
 	DefaultBufferUsageLimit = 0.8
 
 	UDFApplierMessageKey = "x-numa-message-key" // The key in the UDF applier HTTP header used to pass the map-reduce key
+
+	// processing rate and pending prometheus metrics and labels
+	MetricPipelineLabel   = "pipeline"
+	MetricVertexLabel     = "vertex"
+	MetricPeriodLabel     = "period"
+	VertexProcessingRate  = "vertex_processing_rate"
+	VertexPendingMessages = "vertex_pending_messages"
 )
 
 type ContentType string
diff --git a/pkg/apis/proto/daemon/daemon.pb.go b/pkg/apis/proto/daemon/daemon.pb.go
index a54f958e..c1ef6065 100644
--- a/pkg/apis/proto/daemon/daemon.pb.go
+++ b/pkg/apis/proto/daemon/daemon.pb.go
@@ -160,6 +160,70 @@ func (m *BufferInfo) GetIsFull() bool {
 	return false
 }
 
+// VertexMetrics is used to provide information about the vertex including processing rate.
+type VertexMetrics struct {
+	Pipeline             *string            `protobuf:"bytes,1,req,name=pipeline" json:"pipeline,omitempty"`
+	Vertex               *string            `protobuf:"bytes,2,req,name=vertex" json:"vertex,omitempty"`
+	ProcessingRates      map[string]float32 `protobuf:"bytes,3,rep,name=processingRates" json:"processingRates,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"fixed32,2,opt,name=value"`
+	XXX_NoUnkeyedLiteral struct{}           `json:"-"`
+	XXX_unrecognized     []byte             `json:"-"`
+	XXX_sizecache        int32              `json:"-"`
+}
+
+func (m *VertexMetrics) Reset()         { *m = VertexMetrics{} }
+func (m *VertexMetrics) String() string { return proto.CompactTextString(m) }
+func (*VertexMetrics) ProtoMessage()    {}
+func (*VertexMetrics) Descriptor() ([]byte, []int) {
+	return fileDescriptor_93e327fd0d673221, []int{1}
+}
+func (m *VertexMetrics) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *VertexMetrics) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_VertexMetrics.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *VertexMetrics) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_VertexMetrics.Merge(m, src)
+}
+func (m *VertexMetrics) XXX_Size() int {
+	return m.Size()
+}
+func (m *VertexMetrics) XXX_DiscardUnknown() {
+	xxx_messageInfo_VertexMetrics.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_VertexMetrics proto.InternalMessageInfo
+
+func (m *VertexMetrics) GetPipeline() string {
+	if m != nil && m.Pipeline != nil {
+		return *m.Pipeline
+	}
+	return ""
+}
+
+func (m *VertexMetrics) GetVertex() string {
+	if m != nil && m.Vertex != nil {
+		return *m.Vertex
+	}
+	return ""
+}
+
+func (m *VertexMetrics) GetProcessingRates() map[string]float32 {
+	if m != nil {
+		return m.ProcessingRates
+	}
+	return nil
+}
+
 type ListBuffersRequest struct {
 	Pipeline             *string  `protobuf:"bytes,1,req,name=pipeline" json:"pipeline,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
@@ -171,7 +235,7 @@ func (m *ListBuffersRequest) Reset()         { *m = ListBuffersRequest{} }
 func (m *ListBuffersRequest) String() string { return proto.CompactTextString(m) }
 func (*ListBuffersRequest) ProtoMessage()    {}
 func (*ListBuffersRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_93e327fd0d673221, []int{1}
+	return fileDescriptor_93e327fd0d673221, []int{2}
 }
 func (m *ListBuffersRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -218,7 +282,7 @@ func (m *ListBuffersResponse) Reset()         { *m = ListBuffersResponse{} }
 func (m *ListBuffersResponse) String() string { return proto.CompactTextString(m) }
 func (*ListBuffersResponse) ProtoMessage()    {}
 func (*ListBuffersResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_93e327fd0d673221, []int{2}
+	return fileDescriptor_93e327fd0d673221, []int{3}
 }
 func (m *ListBuffersResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -266,7 +330,7 @@ func (m *GetBufferRequest) Reset()         { *m = GetBufferRequest{} }
 func (m *GetBufferRequest) String() string { return proto.CompactTextString(m) }
 func (*GetBufferRequest) ProtoMessage()    {}
 func (*GetBufferRequest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_93e327fd0d673221, []int{3}
+	return fileDescriptor_93e327fd0d673221, []int{4}
 }
 func (m *GetBufferRequest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -320,7 +384,7 @@ func (m *GetBufferResponse) Reset()         { *m = GetBufferResponse{} }
 func (m *GetBufferResponse) String() string { return proto.CompactTextString(m) }
 func (*GetBufferResponse) ProtoMessage()    {}
 func (*GetBufferResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_93e327fd0d673221, []int{4}
+	return fileDescriptor_93e327fd0d673221, []int{5}
 }
 func (m *GetBufferResponse) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -356,12 +420,126 @@ func (m *GetBufferResponse) GetBuffer() *BufferInfo {
 	return nil
 }
 
+type GetVertexMetricsRequest struct {
+	Namespace            *string  `protobuf:"bytes,1,req,name=namespace" json:"namespace,omitempty"`
+	Pipeline             *string  `protobuf:"bytes,2,req,name=pipeline" json:"pipeline,omitempty"`
+	Vertex               *string  `protobuf:"bytes,3,req,name=vertex" json:"vertex,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *GetVertexMetricsRequest) Reset()         { *m = GetVertexMetricsRequest{} }
+func (m *GetVertexMetricsRequest) String() string { return proto.CompactTextString(m) }
+func (*GetVertexMetricsRequest) ProtoMessage()    {}
+func (*GetVertexMetricsRequest) Descriptor() ([]byte, []int) {
+	return fileDescriptor_93e327fd0d673221, []int{6}
+}
+func (m *GetVertexMetricsRequest) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *GetVertexMetricsRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_GetVertexMetricsRequest.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *GetVertexMetricsRequest) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_GetVertexMetricsRequest.Merge(m, src)
+}
+func (m *GetVertexMetricsRequest) XXX_Size() int {
+	return m.Size()
+}
+func (m *GetVertexMetricsRequest) XXX_DiscardUnknown() {
+	xxx_messageInfo_GetVertexMetricsRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_GetVertexMetricsRequest proto.InternalMessageInfo
+
+func (m *GetVertexMetricsRequest) GetNamespace() string {
+	if m != nil && m.Namespace != nil {
+		return *m.Namespace
+	}
+	return ""
+}
+
+func (m *GetVertexMetricsRequest) GetPipeline() string {
+	if m != nil && m.Pipeline != nil {
+		return *m.Pipeline
+	}
+	return ""
+}
+
+func (m *GetVertexMetricsRequest) GetVertex() string {
+	if m != nil && m.Vertex != nil {
+		return *m.Vertex
+	}
+	return ""
+}
+
+type GetVertexMetricsResponse struct {
+	Vertex               *VertexMetrics `protobuf:"bytes,1,req,name=vertex" json:"vertex,omitempty"`
+	XXX_NoUnkeyedLiteral struct{}       `json:"-"`
+	XXX_unrecognized     []byte         `json:"-"`
+	XXX_sizecache        int32          `json:"-"`
+}
+
+func (m *GetVertexMetricsResponse) Reset()         { *m = GetVertexMetricsResponse{} }
+func (m *GetVertexMetricsResponse) String() string { return proto.CompactTextString(m) }
+func (*GetVertexMetricsResponse) ProtoMessage()    {}
+func (*GetVertexMetricsResponse) Descriptor() ([]byte, []int) {
+	return fileDescriptor_93e327fd0d673221, []int{7}
+}
+func (m *GetVertexMetricsResponse) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *GetVertexMetricsResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_GetVertexMetricsResponse.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *GetVertexMetricsResponse) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_GetVertexMetricsResponse.Merge(m, src)
+}
+func (m *GetVertexMetricsResponse) XXX_Size() int {
+	return m.Size()
+}
+func (m *GetVertexMetricsResponse) XXX_DiscardUnknown() {
+	xxx_messageInfo_GetVertexMetricsResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_GetVertexMetricsResponse proto.InternalMessageInfo
+
+func (m *GetVertexMetricsResponse) GetVertex() *VertexMetrics {
+	if m != nil {
+		return m.Vertex
+	}
+	return nil
+}
+
 func init() {
 	proto.RegisterType((*BufferInfo)(nil), "daemon.BufferInfo")
+	proto.RegisterType((*VertexMetrics)(nil), "daemon.VertexMetrics")
+	proto.RegisterMapType((map[string]float32)(nil), "daemon.VertexMetrics.ProcessingRatesEntry")
 	proto.RegisterType((*ListBuffersRequest)(nil), "daemon.ListBuffersRequest")
 	proto.RegisterType((*ListBuffersResponse)(nil), "daemon.ListBuffersResponse")
 	proto.RegisterType((*GetBufferRequest)(nil), "daemon.GetBufferRequest")
 	proto.RegisterType((*GetBufferResponse)(nil), "daemon.GetBufferResponse")
+	proto.RegisterType((*GetVertexMetricsRequest)(nil), "daemon.GetVertexMetricsRequest")
+	proto.RegisterType((*GetVertexMetricsResponse)(nil), "daemon.GetVertexMetricsResponse")
 }
 
 func init() {
@@ -369,39 +547,52 @@ func init() {
 }
 
 var fileDescriptor_93e327fd0d673221 = []byte{
-	// 511 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x54, 0x6f, 0x6b, 0xd3, 0x40,
-	0x18, 0x27, 0xa9, 0x76, 0xed, 0x53, 0x87, 0xf3, 0x04, 0x39, 0x33, 0x29, 0x21, 0x0c, 0x09, 0x63,
-	0xf6, 0xb4, 0xe0, 0x6b, 0x65, 0x93, 0x89, 0x50, 0x45, 0x22, 0xfa, 0xc2, 0x77, 0x69, 0x77, 0xc9,
-	0xce, 0x25, 0x77, 0x31, 0x77, 0xd9, 0x94, 0xb1, 0x37, 0xfb, 0x0a, 0x7e, 0x29, 0x5f, 0x0a, 0x7e,
-	0x01, 0x29, 0x7e, 0x0e, 0x91, 0xdc, 0x5d, 0xba, 0x74, 0x2b, 0x63, 0xaf, 0x7a, 0xcf, 0xef, 0xcf,
-	0xf3, 0x7b, 0xee, 0x9e, 0xb6, 0x10, 0x14, 0x47, 0x29, 0x89, 0x0b, 0x26, 0x49, 0x51, 0x0a, 0x25,
-	0xc8, 0x41, 0x4c, 0x73, 0xc1, 0xed, 0xc7, 0x48, 0x63, 0xa8, 0x6b, 0x2a, 0xef, 0x51, 0x2a, 0x44,
-	0x9a, 0xd1, 0x5a, 0x4e, 0x62, 0xce, 0x85, 0x8a, 0x15, 0x13, 0x5c, 0x1a, 0x95, 0xb7, 0x69, 0x59,
-	0x5d, 0x4d, 0xab, 0x84, 0xd0, 0xbc, 0x50, 0xdf, 0x0d, 0x19, 0x9c, 0x77, 0x00, 0x76, 0xab, 0x24,
-	0xa1, 0xe5, 0x1b, 0x9e, 0x08, 0xe4, 0x41, 0xaf, 0x60, 0x05, 0xcd, 0x18, 0xa7, 0xd8, 0xf1, 0xdd,
-	0xb0, 0x1f, 0x2d, 0x6a, 0x34, 0x04, 0x48, 0x4a, 0x91, 0x7f, 0xa2, 0xa5, 0xa2, 0xdf, 0xb0, 0xab,
-	0xd9, 0x16, 0x52, 0x7b, 0x95, 0xb0, 0x6c, 0xc7, 0x78, 0x9b, 0xba, 0xf6, 0x4e, 0x75, 0xca, 0xbb,
-	0x38, 0xa7, 0xf8, 0x96, 0xf1, 0x5e, 0x20, 0x28, 0x80, 0x3b, 0x05, 0xe5, 0x07, 0x8c, 0xa7, 0x7b,
-	0xa2, 0xe2, 0x0a, 0xdf, 0xf6, 0xdd, 0xb0, 0x13, 0x2d, 0x61, 0x28, 0x84, 0xbb, 0xf1, 0xec, 0xe8,
-	0x7d, 0x5b, 0xd6, 0xd5, 0xb2, 0xcb, 0x30, 0xda, 0x82, 0x75, 0x25, 0x54, 0x9c, 0xbd, 0xa5, 0x52,
-	0xc6, 0x29, 0x95, 0x78, 0x4d, 0xeb, 0x96, 0xc1, 0x3a, 0xd3, 0x4c, 0x30, 0xa1, 0x3c, 0x55, 0x87,
-	0xb8, 0x67, 0x32, 0xdb, 0x18, 0xda, 0x86, 0x0d, 0x53, 0x7f, 0xac, 0x3d, 0x13, 0x96, 0x33, 0x85,
-	0xfb, 0xbe, 0x1b, 0x3a, 0xd1, 0x15, 0x1c, 0xf9, 0x30, 0x68, 0x61, 0x18, 0xb4, 0xac, 0x0d, 0xa1,
-	0x07, 0xd0, 0x65, 0x72, 0xbf, 0xca, 0x32, 0x3c, 0xf0, 0xdd, 0xb0, 0x17, 0xd9, 0x2a, 0x78, 0x0a,
-	0x68, 0xc2, 0xa4, 0x32, 0x7b, 0x90, 0x11, 0xfd, 0x5a, 0x51, 0xa9, 0xae, 0xdb, 0x45, 0xb0, 0x07,
-	0xf7, 0x97, 0x1c, 0xb2, 0x10, 0x5c, 0x52, 0xb4, 0x03, 0x6b, 0x26, 0x4f, 0x62, 0xc7, 0xef, 0x84,
-	0x83, 0x31, 0x1a, 0xd9, 0x2f, 0xcc, 0xc5, 0x8e, 0xa3, 0x46, 0x12, 0xec, 0xc3, 0xc6, 0x6b, 0x6a,
-	0x7b, 0xdc, 0x20, 0xb4, 0x1e, 0xdf, 0x58, 0xed, 0xf2, 0x6d, 0x15, 0xbc, 0x80, 0x7b, 0xad, 0x3e,
-	0x76, 0x94, 0xed, 0x85, 0xb8, 0x6e, 0xb3, 0x7a, 0x12, 0xab, 0x18, 0xff, 0x73, 0x60, 0xfd, 0x95,
-	0x66, 0x3f, 0xd0, 0xf2, 0x98, 0xcd, 0x28, 0x52, 0x30, 0x68, 0xdd, 0x0f, 0x79, 0x8d, 0xf9, 0xea,
-	0x33, 0x79, 0x9b, 0x2b, 0x39, 0x33, 0x45, 0xb0, 0x73, 0xfe, 0xfb, 0xef, 0x0f, 0xf7, 0x31, 0xda,
-	0xd2, 0xbf, 0x8d, 0xe3, 0x67, 0xa4, 0xb9, 0x8c, 0x24, 0xa7, 0xcd, 0xf1, 0x8c, 0xd8, 0x07, 0x41,
-	0x27, 0xd0, 0x5f, 0x5c, 0x04, 0xe1, 0xa6, 0xef, 0xe5, 0x37, 0xf2, 0x1e, 0xae, 0x60, 0x6c, 0xde,
-	0x73, 0x9d, 0x47, 0xd0, 0x93, 0x9b, 0xe4, 0x91, 0x53, 0x73, 0x38, 0xdb, 0x7d, 0xf9, 0x73, 0x3e,
-	0x74, 0x7e, 0xcd, 0x87, 0xce, 0x9f, 0xf9, 0xd0, 0xf9, 0x3c, 0x4e, 0x99, 0x3a, 0xac, 0xa6, 0xa3,
-	0x99, 0xc8, 0x09, 0xaf, 0xf2, 0xb8, 0x28, 0xc5, 0x17, 0x7d, 0x48, 0x32, 0x71, 0x42, 0x56, 0xfe,
-	0x2f, 0xfc, 0x0f, 0x00, 0x00, 0xff, 0xff, 0x7c, 0x49, 0x28, 0x49, 0x2f, 0x04, 0x00, 0x00,
+	// 705 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x8c, 0x55, 0xc1, 0x4e, 0xdb, 0x4a,
+	0x14, 0x95, 0xed, 0x47, 0x20, 0x37, 0x0f, 0xc1, 0x9b, 0xc7, 0xe3, 0x4d, 0x0d, 0x4a, 0x2d, 0x0b,
+	0x55, 0x16, 0x82, 0xb8, 0x45, 0xaa, 0x54, 0x75, 0xd3, 0x0a, 0x5a, 0x10, 0x52, 0x40, 0xc8, 0xa5,
+	0x5d, 0x74, 0x67, 0xc2, 0xc4, 0xb8, 0x89, 0x67, 0x5c, 0xcf, 0x38, 0x14, 0xa1, 0x6c, 0xf8, 0x85,
+	0x7e, 0x46, 0xfb, 0x21, 0x5d, 0x56, 0xea, 0xba, 0x52, 0x85, 0xfa, 0x21, 0x95, 0x67, 0xec, 0x60,
+	0x83, 0x41, 0xac, 0x32, 0xf7, 0xdc, 0x73, 0xef, 0x3d, 0x33, 0xe7, 0x26, 0x01, 0x3b, 0x1e, 0x04,
+	0xae, 0x1f, 0x87, 0xdc, 0x8d, 0x13, 0x26, 0x98, 0x7b, 0xec, 0x93, 0x88, 0xd1, 0xfc, 0xa3, 0x23,
+	0x31, 0xd4, 0x50, 0x91, 0xb9, 0x1c, 0x30, 0x16, 0x0c, 0x49, 0x46, 0x77, 0x7d, 0x4a, 0x99, 0xf0,
+	0x45, 0xc8, 0x28, 0x57, 0x2c, 0x73, 0x29, 0xcf, 0xca, 0xe8, 0x28, 0xed, 0xbb, 0x24, 0x8a, 0xc5,
+	0x99, 0x4a, 0xda, 0x17, 0x06, 0xc0, 0x66, 0xda, 0xef, 0x93, 0x64, 0x97, 0xf6, 0x19, 0x32, 0x61,
+	0x26, 0x0e, 0x63, 0x32, 0x0c, 0x29, 0xc1, 0x9a, 0xa5, 0x3b, 0x4d, 0x6f, 0x12, 0xa3, 0x36, 0x40,
+	0x3f, 0x61, 0xd1, 0x3b, 0x92, 0x08, 0xf2, 0x09, 0xeb, 0x32, 0x5b, 0x42, 0xb2, 0x5a, 0xc1, 0xf2,
+	0xac, 0xa1, 0x6a, 0x8b, 0x38, 0xab, 0x3d, 0x92, 0x53, 0xf6, 0xfd, 0x88, 0xe0, 0xbf, 0x54, 0xed,
+	0x15, 0x82, 0x6c, 0xf8, 0x3b, 0x26, 0xf4, 0x38, 0xa4, 0xc1, 0x16, 0x4b, 0xa9, 0xc0, 0x53, 0x96,
+	0xee, 0x18, 0x5e, 0x05, 0x43, 0x0e, 0xcc, 0xf9, 0xbd, 0xc1, 0x41, 0x99, 0xd6, 0x90, 0xb4, 0xeb,
+	0x30, 0x5a, 0x81, 0x59, 0xc1, 0x84, 0x3f, 0xdc, 0x23, 0x9c, 0xfb, 0x01, 0xe1, 0x78, 0x5a, 0xf2,
+	0xaa, 0x60, 0x36, 0x53, 0x29, 0xe8, 0x12, 0x1a, 0x88, 0x13, 0x3c, 0xa3, 0x66, 0x96, 0x31, 0xb4,
+	0x0a, 0xf3, 0x2a, 0x7e, 0x9b, 0xd5, 0x74, 0xc3, 0x28, 0x14, 0xb8, 0x69, 0xe9, 0x8e, 0xe6, 0xdd,
+	0xc0, 0x91, 0x05, 0xad, 0x12, 0x86, 0x41, 0xd2, 0xca, 0x10, 0x5a, 0x84, 0x46, 0xc8, 0xb7, 0xd3,
+	0xe1, 0x10, 0xb7, 0x2c, 0xdd, 0x99, 0xf1, 0xf2, 0xc8, 0xfe, 0xa9, 0xc1, 0xac, 0x7a, 0xa8, 0x3d,
+	0x22, 0x92, 0xb0, 0xc7, 0xef, 0xf4, 0x61, 0x11, 0x1a, 0xa3, 0xb2, 0x07, 0x79, 0x84, 0x0e, 0x61,
+	0x2e, 0x4e, 0x58, 0x8f, 0x70, 0x1e, 0xd2, 0xc0, 0xf3, 0x05, 0xe1, 0xd8, 0xb0, 0x0c, 0xa7, 0xb5,
+	0xb1, 0xda, 0xc9, 0xb7, 0xa6, 0x32, 0xa3, 0x73, 0x50, 0x25, 0xbf, 0xa6, 0x22, 0x39, 0xf3, 0xae,
+	0xb7, 0x30, 0x37, 0x61, 0xa1, 0x8e, 0x88, 0xe6, 0xc1, 0x18, 0x90, 0x33, 0xac, 0x59, 0x9a, 0xd3,
+	0xf4, 0xb2, 0x23, 0x5a, 0x80, 0xa9, 0x91, 0x3f, 0x4c, 0x09, 0xd6, 0x2d, 0xcd, 0xd1, 0x3d, 0x15,
+	0x3c, 0xd7, 0x9f, 0x69, 0xf6, 0x63, 0x40, 0xdd, 0x90, 0x0b, 0xb5, 0x67, 0xdc, 0x23, 0x1f, 0x53,
+	0xc2, 0xc5, 0x5d, 0x77, 0xb4, 0xb7, 0xe0, 0xdf, 0x4a, 0x05, 0x8f, 0x19, 0xe5, 0x04, 0xad, 0xc1,
+	0xb4, 0x7a, 0x4f, 0x8e, 0x35, 0x79, 0x35, 0x54, 0x5c, 0xed, 0x6a, 0x87, 0xbd, 0x82, 0x62, 0x6f,
+	0xc3, 0xfc, 0x0e, 0xc9, 0x7b, 0xdc, 0x63, 0x68, 0xf6, 0xb0, 0xaa, 0xb4, 0x78, 0x58, 0x15, 0xd9,
+	0x2f, 0xe0, 0x9f, 0x52, 0x9f, 0x5c, 0xca, 0xea, 0x84, 0x9c, 0xb5, 0xa9, 0x57, 0x52, 0x34, 0x18,
+	0xc0, 0xff, 0x3b, 0x44, 0x54, 0x5e, 0xbf, 0xd0, 0xb3, 0x0c, 0x4d, 0xea, 0x47, 0x84, 0xc7, 0x7e,
+	0xaf, 0x10, 0x74, 0x05, 0x54, 0xd4, 0xea, 0xb7, 0xae, 0x81, 0x51, 0x5e, 0x03, 0x7b, 0x17, 0xf0,
+	0xcd, 0x61, 0xb9, 0xe8, 0xf5, 0x49, 0x8d, 0x12, 0xfd, 0x5f, 0xed, 0x66, 0x14, 0xad, 0x36, 0xbe,
+	0x1a, 0x30, 0xfb, 0x4a, 0x12, 0xde, 0x90, 0x64, 0x14, 0xf6, 0x08, 0x12, 0xd0, 0x2a, 0xf9, 0x82,
+	0xcc, 0xa2, 0xfe, 0xa6, 0xbd, 0xe6, 0x52, 0x6d, 0x4e, 0x09, 0xb1, 0xd7, 0x2e, 0x7e, 0xfc, 0xfe,
+	0xac, 0x3f, 0x42, 0x2b, 0xf2, 0x37, 0x6b, 0xf4, 0xc4, 0x2d, 0xae, 0xc5, 0xdd, 0xf3, 0xe2, 0x38,
+	0x76, 0x73, 0x23, 0xd1, 0x29, 0x34, 0x27, 0x06, 0x20, 0x5c, 0xf4, 0xbd, 0xee, 0xad, 0xf9, 0xa0,
+	0x26, 0x93, 0xcf, 0x7b, 0x2a, 0xe7, 0xb9, 0x68, 0xfd, 0x3e, 0xf3, 0xdc, 0x73, 0x75, 0x18, 0xa3,
+	0x2f, 0x9a, 0x5c, 0xa1, 0xea, 0x77, 0xf3, 0x61, 0x69, 0x4c, 0x9d, 0xa7, 0xa6, 0x75, 0x3b, 0x21,
+	0x97, 0x73, 0x28, 0xe5, 0xec, 0xa3, 0x6e, 0x21, 0x67, 0x62, 0xb9, 0x7b, 0x3e, 0x39, 0x8e, 0xeb,
+	0x45, 0x66, 0xde, 0x84, 0xbd, 0x0c, 0x53, 0x2e, 0x8d, 0xdd, 0x48, 0x75, 0xdf, 0x7c, 0xf9, 0xed,
+	0xb2, 0xad, 0x7d, 0xbf, 0x6c, 0x6b, 0xbf, 0x2e, 0xdb, 0xda, 0xfb, 0x8d, 0x20, 0x14, 0x27, 0xe9,
+	0x51, 0xa7, 0xc7, 0x22, 0x97, 0xa6, 0x91, 0x1f, 0x27, 0xec, 0x83, 0x3c, 0xf4, 0x87, 0xec, 0xd4,
+	0xad, 0xfd, 0x77, 0xf9, 0x13, 0x00, 0x00, 0xff, 0xff, 0x82, 0xa0, 0x01, 0xc4, 0x75, 0x06, 0x00,
+	0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -418,6 +609,7 @@ const _ = grpc.SupportPackageIsVersion4
 type DaemonServiceClient interface {
 	ListBuffers(ctx context.Context, in *ListBuffersRequest, opts ...grpc.CallOption) (*ListBuffersResponse, error)
 	GetBuffer(ctx context.Context, in *GetBufferRequest, opts ...grpc.CallOption) (*GetBufferResponse, error)
+	GetVertexMetrics(ctx context.Context, in *GetVertexMetricsRequest, opts ...grpc.CallOption) (*GetVertexMetricsResponse, error)
 }
 
 type daemonServiceClient struct {
@@ -446,10 +638,20 @@ func (c *daemonServiceClient) GetBuffer(ctx context.Context, in *GetBufferReques
 	return out, nil
 }
 
+func (c *daemonServiceClient) GetVertexMetrics(ctx context.Context, in *GetVertexMetricsRequest, opts ...grpc.CallOption) (*GetVertexMetricsResponse, error) {
+	out := new(GetVertexMetricsResponse)
+	err := c.cc.Invoke(ctx, "/daemon.DaemonService/GetVertexMetrics", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // DaemonServiceServer is the server API for DaemonService service.
 type DaemonServiceServer interface {
 	ListBuffers(context.Context, *ListBuffersRequest) (*ListBuffersResponse, error)
 	GetBuffer(context.Context, *GetBufferRequest) (*GetBufferResponse, error)
+	GetVertexMetrics(context.Context, *GetVertexMetricsRequest) (*GetVertexMetricsResponse, error)
 }
 
 // UnimplementedDaemonServiceServer can be embedded to have forward compatible implementations.
@@ -462,6 +664,9 @@ func (*UnimplementedDaemonServiceServer) ListBuffers(ctx context.Context, req *L
 func (*UnimplementedDaemonServiceServer) GetBuffer(ctx context.Context, req *GetBufferRequest) (*GetBufferResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetBuffer not implemented")
 }
+func (*UnimplementedDaemonServiceServer) GetVertexMetrics(ctx context.Context, req *GetVertexMetricsRequest) (*GetVertexMetricsResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetVertexMetrics not implemented")
+}
 
 func RegisterDaemonServiceServer(s *grpc.Server, srv DaemonServiceServer) {
 	s.RegisterService(&_DaemonService_serviceDesc, srv)
@@ -503,6 +708,24 @@ func _DaemonService_GetBuffer_Handler(srv interface{}, ctx context.Context, dec
 	return interceptor(ctx, in, info, handler)
 }
 
+func _DaemonService_GetVertexMetrics_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(GetVertexMetricsRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(DaemonServiceServer).GetVertexMetrics(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/daemon.DaemonService/GetVertexMetrics",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(DaemonServiceServer).GetVertexMetrics(ctx, req.(*GetVertexMetricsRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 var _DaemonService_serviceDesc = grpc.ServiceDesc{
 	ServiceName: "daemon.DaemonService",
 	HandlerType: (*DaemonServiceServer)(nil),
@@ -515,6 +738,10 @@ var _DaemonService_serviceDesc = grpc.ServiceDesc{
 			MethodName: "GetBuffer",
 			Handler:    _DaemonService_GetBuffer_Handler,
 		},
+		{
+			MethodName: "GetVertexMetrics",
+			Handler:    _DaemonService_GetVertexMetrics_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "pkg/apis/proto/daemon/daemon.proto",
@@ -639,6 +866,69 @@ func (m *BufferInfo) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *VertexMetrics) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *VertexMetrics) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *VertexMetrics) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.ProcessingRates) > 0 {
+		for k := range m.ProcessingRates {
+			v := m.ProcessingRates[k]
+			baseI := i
+			i -= 4
+			encoding_binary.LittleEndian.PutUint32(dAtA[i:], uint32(math.Float32bits(float32(v))))
+			i--
+			dAtA[i] = 0x15
+			i -= len(k)
+			copy(dAtA[i:], k)
+			i = encodeVarintDaemon(dAtA, i, uint64(len(k)))
+			i--
+			dAtA[i] = 0xa
+			i = encodeVarintDaemon(dAtA, i, uint64(baseI-i))
+			i--
+			dAtA[i] = 0x1a
+		}
+	}
+	if m.Vertex == nil {
+		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("vertex")
+	} else {
+		i -= len(*m.Vertex)
+		copy(dAtA[i:], *m.Vertex)
+		i = encodeVarintDaemon(dAtA, i, uint64(len(*m.Vertex)))
+		i--
+		dAtA[i] = 0x12
+	}
+	if m.Pipeline == nil {
+		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("pipeline")
+	} else {
+		i -= len(*m.Pipeline)
+		copy(dAtA[i:], *m.Pipeline)
+		i = encodeVarintDaemon(dAtA, i, uint64(len(*m.Pipeline)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
 func (m *ListBuffersRequest) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -802,6 +1092,101 @@ func (m *GetBufferResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
+func (m *GetVertexMetricsRequest) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *GetVertexMetricsRequest) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *GetVertexMetricsRequest) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if m.Vertex == nil {
+		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("vertex")
+	} else {
+		i -= len(*m.Vertex)
+		copy(dAtA[i:], *m.Vertex)
+		i = encodeVarintDaemon(dAtA, i, uint64(len(*m.Vertex)))
+		i--
+		dAtA[i] = 0x1a
+	}
+	if m.Pipeline == nil {
+		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("pipeline")
+	} else {
+		i -= len(*m.Pipeline)
+		copy(dAtA[i:], *m.Pipeline)
+		i = encodeVarintDaemon(dAtA, i, uint64(len(*m.Pipeline)))
+		i--
+		dAtA[i] = 0x12
+	}
+	if m.Namespace == nil {
+		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("namespace")
+	} else {
+		i -= len(*m.Namespace)
+		copy(dAtA[i:], *m.Namespace)
+		i = encodeVarintDaemon(dAtA, i, uint64(len(*m.Namespace)))
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *GetVertexMetricsResponse) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *GetVertexMetricsResponse) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *GetVertexMetricsResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if m.Vertex == nil {
+		return 0, github_com_gogo_protobuf_proto.NewRequiredNotSetError("vertex")
+	} else {
+		{
+			size, err := m.Vertex.MarshalToSizedBuffer(dAtA[:i])
+			if err != nil {
+				return 0, err
+			}
+			i -= size
+			i = encodeVarintDaemon(dAtA, i, uint64(size))
+		}
+		i--
+		dAtA[i] = 0xa
+	}
+	return len(dAtA) - i, nil
+}
+
 func encodeVarintDaemon(dAtA []byte, offset int, v uint64) int {
 	offset -= sovDaemon(v)
 	base := offset
@@ -862,7 +1247,7 @@ func (m *BufferInfo) Size() (n int) {
 	return n
 }
 
-func (m *ListBuffersRequest) Size() (n int) {
+func (m *VertexMetrics) Size() (n int) {
 	if m == nil {
 		return 0
 	}
@@ -872,20 +1257,48 @@ func (m *ListBuffersRequest) Size() (n int) {
 		l = len(*m.Pipeline)
 		n += 1 + l + sovDaemon(uint64(l))
 	}
+	if m.Vertex != nil {
+		l = len(*m.Vertex)
+		n += 1 + l + sovDaemon(uint64(l))
+	}
+	if len(m.ProcessingRates) > 0 {
+		for k, v := range m.ProcessingRates {
+			_ = k
+			_ = v
+			mapEntrySize := 1 + len(k) + sovDaemon(uint64(len(k))) + 1 + 4
+			n += mapEntrySize + 1 + sovDaemon(uint64(mapEntrySize))
+		}
+	}
 	if m.XXX_unrecognized != nil {
 		n += len(m.XXX_unrecognized)
 	}
 	return n
 }
 
-func (m *ListBuffersResponse) Size() (n int) {
+func (m *ListBuffersRequest) Size() (n int) {
 	if m == nil {
 		return 0
 	}
 	var l int
 	_ = l
-	if len(m.Buffers) > 0 {
-		for _, e := range m.Buffers {
+	if m.Pipeline != nil {
+		l = len(*m.Pipeline)
+		n += 1 + l + sovDaemon(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *ListBuffersResponse) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if len(m.Buffers) > 0 {
+		for _, e := range m.Buffers {
 			l = e.Size()
 			n += 1 + l + sovDaemon(uint64(l))
 		}
@@ -932,6 +1345,46 @@ func (m *GetBufferResponse) Size() (n int) {
 	return n
 }
 
+func (m *GetVertexMetricsRequest) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Namespace != nil {
+		l = len(*m.Namespace)
+		n += 1 + l + sovDaemon(uint64(l))
+	}
+	if m.Pipeline != nil {
+		l = len(*m.Pipeline)
+		n += 1 + l + sovDaemon(uint64(l))
+	}
+	if m.Vertex != nil {
+		l = len(*m.Vertex)
+		n += 1 + l + sovDaemon(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *GetVertexMetricsResponse) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.Vertex != nil {
+		l = m.Vertex.Size()
+		n += 1 + l + sovDaemon(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func sovDaemon(x uint64) (n int) {
 	return (math_bits.Len64(x|1) + 6) / 7
 }
@@ -1291,6 +1744,238 @@ func (m *BufferInfo) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *VertexMetrics) Unmarshal(dAtA []byte) error {
+	var hasFields [1]uint64
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowDaemon
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: VertexMetrics: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: VertexMetrics: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Pipeline", 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.Pipeline = &s
+			iNdEx = postIndex
+			hasFields[0] |= uint64(0x00000001)
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Vertex", 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.Vertex = &s
+			iNdEx = postIndex
+			hasFields[0] |= uint64(0x00000002)
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ProcessingRates", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowDaemon
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.ProcessingRates == nil {
+				m.ProcessingRates = make(map[string]float32)
+			}
+			var mapkey string
+			var mapvalue float32
+			for iNdEx < postIndex {
+				entryPreIndex := iNdEx
+				var wire uint64
+				for shift := uint(0); ; shift += 7 {
+					if shift >= 64 {
+						return ErrIntOverflowDaemon
+					}
+					if iNdEx >= l {
+						return io.ErrUnexpectedEOF
+					}
+					b := dAtA[iNdEx]
+					iNdEx++
+					wire |= uint64(b&0x7F) << shift
+					if b < 0x80 {
+						break
+					}
+				}
+				fieldNum := int32(wire >> 3)
+				if fieldNum == 1 {
+					var stringLenmapkey uint64
+					for shift := uint(0); ; shift += 7 {
+						if shift >= 64 {
+							return ErrIntOverflowDaemon
+						}
+						if iNdEx >= l {
+							return io.ErrUnexpectedEOF
+						}
+						b := dAtA[iNdEx]
+						iNdEx++
+						stringLenmapkey |= uint64(b&0x7F) << shift
+						if b < 0x80 {
+							break
+						}
+					}
+					intStringLenmapkey := int(stringLenmapkey)
+					if intStringLenmapkey < 0 {
+						return ErrInvalidLengthDaemon
+					}
+					postStringIndexmapkey := iNdEx + intStringLenmapkey
+					if postStringIndexmapkey < 0 {
+						return ErrInvalidLengthDaemon
+					}
+					if postStringIndexmapkey > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapkey = string(dAtA[iNdEx:postStringIndexmapkey])
+					iNdEx = postStringIndexmapkey
+				} else if fieldNum == 2 {
+					var mapvaluetemp uint32
+					if (iNdEx + 4) > l {
+						return io.ErrUnexpectedEOF
+					}
+					mapvaluetemp = uint32(encoding_binary.LittleEndian.Uint32(dAtA[iNdEx:]))
+					iNdEx += 4
+					mapvalue = math.Float32frombits(mapvaluetemp)
+				} else {
+					iNdEx = entryPreIndex
+					skippy, err := skipDaemon(dAtA[iNdEx:])
+					if err != nil {
+						return err
+					}
+					if (skippy < 0) || (iNdEx+skippy) < 0 {
+						return ErrInvalidLengthDaemon
+					}
+					if (iNdEx + skippy) > postIndex {
+						return io.ErrUnexpectedEOF
+					}
+					iNdEx += skippy
+				}
+			}
+			m.ProcessingRates[mapkey] = mapvalue
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipDaemon(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+	if hasFields[0]&uint64(0x00000001) == 0 {
+		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("pipeline")
+	}
+	if hasFields[0]&uint64(0x00000002) == 0 {
+		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("vertex")
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func (m *ListBuffersRequest) Unmarshal(dAtA []byte) error {
 	var hasFields [1]uint64
 	l := len(dAtA)
@@ -1683,6 +2368,261 @@ func (m *GetBufferResponse) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *GetVertexMetricsRequest) Unmarshal(dAtA []byte) error {
+	var hasFields [1]uint64
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowDaemon
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: GetVertexMetricsRequest: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: GetVertexMetricsRequest: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Namespace", 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.Namespace = &s
+			iNdEx = postIndex
+			hasFields[0] |= uint64(0x00000001)
+		case 2:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Pipeline", 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.Pipeline = &s
+			iNdEx = postIndex
+			hasFields[0] |= uint64(0x00000002)
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Vertex", 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.Vertex = &s
+			iNdEx = postIndex
+			hasFields[0] |= uint64(0x00000004)
+		default:
+			iNdEx = preIndex
+			skippy, err := skipDaemon(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+	if hasFields[0]&uint64(0x00000001) == 0 {
+		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("namespace")
+	}
+	if hasFields[0]&uint64(0x00000002) == 0 {
+		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("pipeline")
+	}
+	if hasFields[0]&uint64(0x00000004) == 0 {
+		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("vertex")
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *GetVertexMetricsResponse) Unmarshal(dAtA []byte) error {
+	var hasFields [1]uint64
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowDaemon
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: GetVertexMetricsResponse: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: GetVertexMetricsResponse: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Vertex", wireType)
+			}
+			var msglen int
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowDaemon
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				msglen |= int(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			if msglen < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			postIndex := iNdEx + msglen
+			if postIndex < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			if m.Vertex == nil {
+				m.Vertex = &VertexMetrics{}
+			}
+			if err := m.Vertex.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
+			}
+			iNdEx = postIndex
+			hasFields[0] |= uint64(0x00000001)
+		default:
+			iNdEx = preIndex
+			skippy, err := skipDaemon(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthDaemon
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+	if hasFields[0]&uint64(0x00000001) == 0 {
+		return github_com_gogo_protobuf_proto.NewRequiredNotSetError("vertex")
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func skipDaemon(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/pkg/apis/proto/daemon/daemon.pb.gw.go b/pkg/apis/proto/daemon/daemon.pb.gw.go
index dc06bfd1..29010afc 100644
--- a/pkg/apis/proto/daemon/daemon.pb.gw.go
+++ b/pkg/apis/proto/daemon/daemon.pb.gw.go
@@ -163,6 +163,104 @@ func local_request_DaemonService_GetBuffer_0(ctx context.Context, marshaler runt
 
 }
 
+func request_DaemonService_GetVertexMetrics_0(ctx context.Context, marshaler runtime.Marshaler, client DaemonServiceClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var protoReq GetVertexMetricsRequest
+	var metadata runtime.ServerMetadata
+
+	var (
+		val string
+		ok  bool
+		err error
+		_   = err
+	)
+
+	val, ok = pathParams["namespace"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "namespace")
+	}
+
+	protoReq.Namespace, err = runtime.StringP(val)
+
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "namespace", err)
+	}
+
+	val, ok = pathParams["pipeline"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "pipeline")
+	}
+
+	protoReq.Pipeline, err = runtime.StringP(val)
+
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "pipeline", err)
+	}
+
+	val, ok = pathParams["vertex"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "vertex")
+	}
+
+	protoReq.Vertex, err = runtime.StringP(val)
+
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "vertex", err)
+	}
+
+	msg, err := client.GetVertexMetrics(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+
+}
+
+func local_request_DaemonService_GetVertexMetrics_0(ctx context.Context, marshaler runtime.Marshaler, server DaemonServiceServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var protoReq GetVertexMetricsRequest
+	var metadata runtime.ServerMetadata
+
+	var (
+		val string
+		ok  bool
+		err error
+		_   = err
+	)
+
+	val, ok = pathParams["namespace"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "namespace")
+	}
+
+	protoReq.Namespace, err = runtime.StringP(val)
+
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "namespace", err)
+	}
+
+	val, ok = pathParams["pipeline"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "pipeline")
+	}
+
+	protoReq.Pipeline, err = runtime.StringP(val)
+
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "pipeline", err)
+	}
+
+	val, ok = pathParams["vertex"]
+	if !ok {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "vertex")
+	}
+
+	protoReq.Vertex, err = runtime.StringP(val)
+
+	if err != nil {
+		return nil, metadata, status.Errorf(codes.InvalidArgument, "type mismatch, parameter: %s, error: %v", "vertex", err)
+	}
+
+	msg, err := server.GetVertexMetrics(ctx, &protoReq)
+	return msg, metadata, err
+
+}
+
 // RegisterDaemonServiceHandlerServer registers the http handlers for service DaemonService to "mux".
 // UnaryRPC     :call DaemonServiceServer directly.
 // StreamingRPC :currently unsupported pending https://github.com/grpc/grpc-go/issues/906.
@@ -215,6 +313,29 @@ func RegisterDaemonServiceHandlerServer(ctx context.Context, mux *runtime.ServeM
 
 	})
 
+	mux.Handle("GET", pattern_DaemonService_GetVertexMetrics_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		rctx, err := runtime.AnnotateIncomingContext(ctx, mux, req)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_DaemonService_GetVertexMetrics_0(rctx, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		ctx = runtime.NewServerMetadataContext(ctx, md)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+
+		forward_DaemonService_GetVertexMetrics_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+	})
+
 	return nil
 }
 
@@ -296,6 +417,26 @@ func RegisterDaemonServiceHandlerClient(ctx context.Context, mux *runtime.ServeM
 
 	})
 
+	mux.Handle("GET", pattern_DaemonService_GetVertexMetrics_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		rctx, err := runtime.AnnotateContext(ctx, mux, req)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_DaemonService_GetVertexMetrics_0(rctx, inboundMarshaler, client, req, pathParams)
+		ctx = runtime.NewServerMetadataContext(ctx, md)
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+
+		forward_DaemonService_GetVertexMetrics_0(ctx, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+	})
+
 	return nil
 }
 
@@ -303,10 +444,14 @@ var (
 	pattern_DaemonService_ListBuffers_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 1, 0, 4, 1, 5, 3, 2, 4}, []string{"api", "v1", "pipelines", "pipeline", "buffers"}, "", runtime.AssumeColonVerbOpt(true)))
 
 	pattern_DaemonService_GetBuffer_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 1, 0, 4, 1, 5, 3, 2, 4, 1, 0, 4, 1, 5, 5}, []string{"api", "v1", "pipelines", "pipeline", "buffers", "buffer"}, "", runtime.AssumeColonVerbOpt(true)))
+
+	pattern_DaemonService_GetVertexMetrics_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2, 1, 0, 4, 1, 5, 2, 2, 3, 1, 0, 4, 1, 5, 4, 2, 5, 1, 0, 4, 1, 5, 6, 2, 7}, []string{"api", "v1", "namespace", "pipelines", "pipeline", "vertices", "vertex", "metrics"}, "", runtime.AssumeColonVerbOpt(true)))
 )
 
 var (
 	forward_DaemonService_ListBuffers_0 = runtime.ForwardResponseMessage
 
 	forward_DaemonService_GetBuffer_0 = runtime.ForwardResponseMessage
+
+	forward_DaemonService_GetVertexMetrics_0 = runtime.ForwardResponseMessage
 )
diff --git a/pkg/apis/proto/daemon/daemon.proto b/pkg/apis/proto/daemon/daemon.proto
index a76b17d9..b17d8fe3 100644
--- a/pkg/apis/proto/daemon/daemon.proto
+++ b/pkg/apis/proto/daemon/daemon.proto
@@ -23,6 +23,13 @@ message BufferInfo {
   required bool isFull = 11;
 }
 
+// VertexMetrics is used to provide information about the vertex including processing rate.
+message VertexMetrics {
+  required string pipeline = 1;
+  required string vertex = 2;
+  map<string, float> processingRates = 3;
+}
+
 message ListBuffersRequest {
     required string pipeline = 1;
 }
@@ -40,6 +47,17 @@ message GetBufferResponse {
     required BufferInfo buffer = 1;
 }
 
+message GetVertexMetricsRequest {
+  required string namespace = 1;
+  required string pipeline = 2;
+  required string vertex = 3;
+
+}
+
+message GetVertexMetricsResponse {
+  required VertexMetrics vertex = 1;
+}
+
 // DaemonService is a grpc service that is used to provide APIs for giving any pipeline information.
 service DaemonService {
 
@@ -50,4 +68,8 @@ service DaemonService {
   rpc GetBuffer (GetBufferRequest) returns (GetBufferResponse) {
     option (google.api.http).get = "/api/v1/pipelines/{pipeline}/buffers/{buffer}";
   };
+
+  rpc GetVertexMetrics (GetVertexMetricsRequest) returns (GetVertexMetricsResponse) {
+    option (google.api.http).get = "/api/v1/namespace/{namespace}/pipelines/{pipeline}/vertices/{vertex}/metrics";
+  };
 }
\ No newline at end of file
diff --git a/pkg/daemon/client/daemon_client.go b/pkg/daemon/client/daemon_client.go
index 72dae562..7345a2a4 100644
--- a/pkg/daemon/client/daemon_client.go
+++ b/pkg/daemon/client/daemon_client.go
@@ -61,3 +61,15 @@ func (dc *DaemonClient) GetPipelineBuffer(ctx context.Context, pipeline, buffer
 		return rspn.Buffer, nil
 	}
 }
+
+func (dc *DaemonClient) GetVertexMetrics(ctx context.Context, namespace, pipeline, vertex string) (*daemon.VertexMetrics, error) {
+	if rspn, err := dc.client.GetVertexMetrics(ctx, &daemon.GetVertexMetricsRequest{
+		Namespace: &namespace,
+		Pipeline:  &pipeline,
+		Vertex:    &vertex,
+	}); err != nil {
+		return nil, err
+	} else {
+		return rspn.Vertex, nil
+	}
+}
diff --git a/pkg/daemon/server/daemon_server.go b/pkg/daemon/server/daemon_server.go
index 75ed5124..2eba43d3 100644
--- a/pkg/daemon/server/daemon_server.go
+++ b/pkg/daemon/server/daemon_server.go
@@ -103,7 +103,7 @@ func (ds *daemonServer) newGRPCServer(isbSvcClient isbsvc.ISBService) *grpc.Serv
 	}
 	grpcServer := grpc.NewServer(sOpts...)
 	grpc_prometheus.Register(grpcServer)
-	daemon.RegisterDaemonServiceServer(grpcServer, service.NewISBSvcQueryService(isbSvcClient, ds.pipeline))
+	daemon.RegisterDaemonServiceServer(grpcServer, service.NewPipelineMetricsQueryService(isbSvcClient, ds.pipeline))
 	return grpcServer
 }
 
diff --git a/pkg/daemon/server/service/isbs_query_service.go b/pkg/daemon/server/service/isbs_query_service.go
deleted file mode 100644
index 70641df4..00000000
--- a/pkg/daemon/server/service/isbs_query_service.go
+++ /dev/null
@@ -1,117 +0,0 @@
-package service
-
-import (
-	"context"
-	"fmt"
-
-	"k8s.io/utils/pointer"
-
-	"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/logging"
-)
-
-type isbSvcQueryService struct {
-	client   isbsvc.ISBService
-	pipeline *v1alpha1.Pipeline
-}
-
-func NewISBSvcQueryService(client isbsvc.ISBService, pipeline *v1alpha1.Pipeline) *isbSvcQueryService {
-	return &isbSvcQueryService{
-		client:   client,
-		pipeline: pipeline,
-	}
-}
-
-// ListBuffers is used to obtain the all the edge buffers information of a pipeline
-func (is *isbSvcQueryService) ListBuffers(ctx context.Context, req *daemon.ListBuffersRequest) (*daemon.ListBuffersResponse, error) {
-	log := logging.FromContext(ctx)
-	resp := new(daemon.ListBuffersResponse)
-
-	buffers := []*daemon.BufferInfo{}
-	for _, edge := range is.pipeline.Spec.Edges {
-		buffer := v1alpha1.GenerateEdgeBufferName(is.pipeline.Namespace, is.pipeline.Name, edge.From, edge.To)
-		bufferInfo, err := is.client.GetBufferInfo(ctx, v1alpha1.Buffer{Name: buffer, Type: v1alpha1.EdgeBuffer})
-		if err != nil {
-			return nil, fmt.Errorf("failed to get information of buffer %q", buffer)
-		}
-		log.Debugf("Buffer %s has bufferInfo %+v", buffer, bufferInfo)
-		bufferLength, bufferUsageLimit := getBufferLimits(is.pipeline, edge)
-		usage := float64(bufferInfo.TotalMessages) / float64(bufferLength)
-		if x := (float64(bufferInfo.PendingCount) + float64(bufferInfo.AckPendingCount)) / float64(bufferLength); x < usage {
-			usage = x
-		}
-		b := &daemon.BufferInfo{
-			Pipeline:         &is.pipeline.Name,
-			FromVertex:       pointer.String(fmt.Sprintf("%v", edge.From)),
-			ToVertex:         pointer.String(fmt.Sprintf("%v", edge.To)),
-			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
-}
-
-// GetBuffer is used to obtain one buffer information of a pipeline
-func (is *isbSvcQueryService) GetBuffer(ctx context.Context, req *daemon.GetBufferRequest) (*daemon.GetBufferResponse, error) {
-	bufferInfo, err := is.client.GetBufferInfo(ctx, v1alpha1.Buffer{Name: *req.Buffer, Type: v1alpha1.EdgeBuffer})
-	if err != nil {
-		return nil, fmt.Errorf("failed to get information of buffer %q", *req.Buffer)
-	}
-	edge := is.pipeline.FindEdgeWithBuffer(*req.Buffer)
-	if edge == nil {
-		return nil, fmt.Errorf("unexpected error, buffer %q not found from the pipeline", *req.Buffer)
-	}
-	bufferLength, bufferUsageLimit := getBufferLimits(is.pipeline, *edge)
-	usage := float64(bufferInfo.TotalMessages) / float64(bufferLength)
-	if x := (float64(bufferInfo.PendingCount) + float64(bufferInfo.AckPendingCount)) / float64(bufferLength); x < usage {
-		usage = x
-	}
-	b := &daemon.BufferInfo{
-		Pipeline:         &is.pipeline.Name,
-		FromVertex:       &edge.From,
-		ToVertex:         &edge.To,
-		BufferName:       req.Buffer,
-		PendingCount:     &bufferInfo.PendingCount,
-		AckPendingCount:  &bufferInfo.AckPendingCount,
-		TotalMessages:    &bufferInfo.TotalMessages,
-		BufferLength:     &bufferLength,
-		BufferUsageLimit: &bufferUsageLimit,
-		BufferUsage:      &usage,
-		IsFull:           pointer.Bool(usage >= bufferUsageLimit),
-	}
-	resp := new(daemon.GetBufferResponse)
-	resp.Buffer = b
-	return resp, nil
-}
-
-func getBufferLimits(pl *v1alpha1.Pipeline, edge v1alpha1.Edge) (bufferLength int64, bufferUsageLimit float64) {
-	bufferLength = int64(v1alpha1.DefaultBufferLength)
-	bufferUsageLimit = v1alpha1.DefaultBufferUsageLimit
-	if x := pl.Spec.Limits; x != nil {
-		if x.BufferMaxLength != nil {
-			bufferLength = int64(*x.BufferMaxLength)
-		}
-		if x.BufferUsageLimit != nil {
-			bufferUsageLimit = float64(*x.BufferUsageLimit) / 100
-		}
-	}
-	if x := edge.Limits; x != nil {
-		if x.BufferMaxLength != nil {
-			bufferLength = int64(*x.BufferMaxLength)
-		}
-		if x.BufferUsageLimit != nil {
-			bufferUsageLimit = float64(*x.BufferUsageLimit) / 100
-		}
-	}
-	return bufferLength, bufferUsageLimit
-}
diff --git a/pkg/daemon/server/service/pipeline_metrics_query_service.go b/pkg/daemon/server/service/pipeline_metrics_query_service.go
new file mode 100644
index 00000000..5b735ff1
--- /dev/null
+++ b/pkg/daemon/server/service/pipeline_metrics_query_service.go
@@ -0,0 +1,192 @@
+package service
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/prometheus/common/expfmt"
+	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"
+	"github.com/numaproj/numaflow/pkg/isbsvc"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+)
+
+// metricsHttpClient interface for the GET call to metrics endpoint.
+// Had to add this an interface for testing
+type metricsHttpClient interface {
+	Get(url string) (*http.Response, error)
+}
+
+type pipelineMetricsQueryService struct {
+	isbsvcClient isbsvc.ISBService
+	pipeline     *v1alpha1.Pipeline
+	httpClient   metricsHttpClient
+}
+
+// NewPipelineMetricsQueryService returns a new instance of pipelineMetricsQueryService
+func NewPipelineMetricsQueryService(isbSvcClient isbsvc.ISBService, pipeline *v1alpha1.Pipeline) *pipelineMetricsQueryService {
+	return &pipelineMetricsQueryService{
+		isbsvcClient: isbSvcClient,
+		pipeline:     pipeline,
+		httpClient: &http.Client{Transport: &http.Transport{
+			TLSHandshakeTimeout: time.Second * 3,
+			TLSClientConfig:     &tls.Config{InsecureSkipVerify: true},
+		}},
+	}
+}
+
+// ListBuffers is used to obtain the all the edge buffers information of a pipeline
+func (ps *pipelineMetricsQueryService) ListBuffers(ctx context.Context, req *daemon.ListBuffersRequest) (*daemon.ListBuffersResponse, error) {
+	log := logging.FromContext(ctx)
+	resp := new(daemon.ListBuffersResponse)
+
+	buffers := []*daemon.BufferInfo{}
+	for _, edge := range ps.pipeline.Spec.Edges {
+		buffer := v1alpha1.GenerateEdgeBufferName(ps.pipeline.Namespace, ps.pipeline.Name, edge.From, edge.To)
+		bufferInfo, err := ps.isbsvcClient.GetBufferInfo(ctx, v1alpha1.Buffer{Name: buffer, Type: v1alpha1.EdgeBuffer})
+		if err != nil {
+			return nil, fmt.Errorf("failed to get information of buffer %q", buffer)
+		}
+		log.Debugf("Buffer %s has bufferInfo %+v", buffer, bufferInfo)
+		bufferLength, bufferUsageLimit := getBufferLimits(ps.pipeline, edge)
+		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,
+			FromVertex:       pointer.String(fmt.Sprintf("%v", edge.From)),
+			ToVertex:         pointer.String(fmt.Sprintf("%v", edge.To)),
+			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
+}
+
+// GetBuffer is used to obtain one buffer information of a pipeline
+func (ps *pipelineMetricsQueryService) GetBuffer(ctx context.Context, req *daemon.GetBufferRequest) (*daemon.GetBufferResponse, error) {
+	bufferInfo, err := ps.isbsvcClient.GetBufferInfo(ctx, v1alpha1.Buffer{Name: *req.Buffer, Type: v1alpha1.EdgeBuffer})
+	if err != nil {
+		return nil, fmt.Errorf("failed to get information of buffer %q:%v", *req.Buffer, err)
+	}
+	edge := ps.pipeline.FindEdgeWithBuffer(*req.Buffer)
+	if edge == nil {
+		return nil, fmt.Errorf("unexpected error, buffer %q not found from the pipeline", *req.Buffer)
+	}
+	bufferLength, bufferUsageLimit := getBufferLimits(ps.pipeline, *edge)
+	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,
+		FromVertex:       &edge.From,
+		ToVertex:         &edge.To,
+		BufferName:       req.Buffer,
+		PendingCount:     &bufferInfo.PendingCount,
+		AckPendingCount:  &bufferInfo.AckPendingCount,
+		TotalMessages:    &bufferInfo.TotalMessages,
+		BufferLength:     &bufferLength,
+		BufferUsageLimit: &bufferUsageLimit,
+		BufferUsage:      &usage,
+		IsFull:           pointer.Bool(usage >= bufferUsageLimit),
+	}
+	resp := new(daemon.GetBufferResponse)
+	resp.Buffer = b
+	return resp, nil
+}
+
+// GetVertexMetrics is used to query the metrics service and is used to obtain the processing rate of a given vertex for 1m, 5m and 15m.
+// In the future maybe latency will also be added here?
+// Should this method live here or maybe another file?
+func (ps *pipelineMetricsQueryService) GetVertexMetrics(ctx context.Context, req *daemon.GetVertexMetricsRequest) (*daemon.GetVertexMetricsResponse, error) {
+	log := logging.FromContext(ctx)
+	resp := new(daemon.GetVertexMetricsResponse)
+
+	vertexName := fmt.Sprintf("%s-%s", ps.pipeline.Name, req.GetVertex())
+	vertex := &v1alpha1.Vertex{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: vertexName,
+		},
+	}
+	// Get the headless service name
+	headlessServiceName := vertex.GetHeadlessServiceName()
+	// We can query the metrics endpoint of the 0th pod to obtain this value.
+	// example: https://simple-pipeline-in-0.simple-pipeline-in-headless.svc.cluster.local:2469/metrics
+	url := fmt.Sprintf("https://%s-0.%s.%s.svc.cluster.local:%v/metrics", vertexName, headlessServiceName, req.GetNamespace(), v1alpha1.VertexMetricsPort)
+
+	res, err := ps.httpClient.Get(url)
+
+	if err != nil {
+		log.Errorf("Error reading the metrics endpoint: %s", err.Error())
+		return nil, err
+	}
+
+	// expfmt Parser from prometheus to parse the metrics
+	textParser := expfmt.TextParser{}
+	result, err := textParser.TextToMetricFamilies(res.Body)
+	if err != nil {
+		log.Errorf("Error in parsing to prometheus metric families: %s", err.Error())
+		return nil, err
+	}
+
+	processingRates := make(map[string]float32, 0)
+	// Check if the resultant metrics list contains the processingRate, if it does look for the period label
+	if value, ok := result[v1alpha1.VertexProcessingRate]; ok {
+		metrics := value.GetMetric()
+		for _, metric := range metrics {
+			labels := metric.GetLabel()
+			for _, label := range labels {
+				if label.GetName() == v1alpha1.MetricPeriodLabel {
+					lookback := label.GetValue()
+					rate := float32(metric.Gauge.GetValue())
+					processingRates[lookback] = rate
+				}
+			}
+		}
+	}
+	v := &daemon.VertexMetrics{
+		Pipeline:        &ps.pipeline.Name,
+		Vertex:          req.Vertex,
+		ProcessingRates: processingRates,
+	}
+	resp.Vertex = v
+	return resp, nil
+}
+
+func getBufferLimits(pl *v1alpha1.Pipeline, edge v1alpha1.Edge) (bufferLength int64, bufferUsageLimit float64) {
+	bufferLength = int64(v1alpha1.DefaultBufferLength)
+	bufferUsageLimit = v1alpha1.DefaultBufferUsageLimit
+	if x := pl.Spec.Limits; x != nil {
+		if x.BufferMaxLength != nil {
+			bufferLength = int64(*x.BufferMaxLength)
+		}
+		if x.BufferUsageLimit != nil {
+			bufferUsageLimit = float64(*x.BufferUsageLimit) / 100
+		}
+	}
+	if x := edge.Limits; x != nil {
+		if x.BufferMaxLength != nil {
+			bufferLength = int64(*x.BufferMaxLength)
+		}
+		if x.BufferUsageLimit != nil {
+			bufferUsageLimit = float64(*x.BufferUsageLimit) / 100
+		}
+	}
+	return bufferLength, bufferUsageLimit
+}
diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go
index 24c4f22e..ccb8d53e 100644
--- a/pkg/metrics/metrics.go
+++ b/pkg/metrics/metrics.go
@@ -9,29 +9,30 @@ import (
 	"os"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"go.uber.org/zap"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
 	sharedtls "github.com/numaproj/numaflow/pkg/shared/tls"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"go.uber.org/zap"
 )
 
 var (
 	processingRate = promauto.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "vertex_processing_rate",
+		Name: dfv1.VertexProcessingRate,
 		Help: "Message processing rate in the last period of seconds, tps. It represents the rate of a vertex instead of a pod.",
-	}, []string{"pipeline", "vertex", "period"})
+	}, []string{dfv1.MetricPipelineLabel, dfv1.MetricVertexLabel, dfv1.MetricPeriodLabel})
 
 	pending = promauto.NewGaugeVec(prometheus.GaugeOpts{
-		Name: "vertex_pending_messages",
+		Name: dfv1.VertexPendingMessages,
 		Help: "Average pending messages in the last period of seconds. It is the pending messages of a vertex, not a pod.",
-	}, []string{"pipeline", "vertex", "period"})
+	}, []string{dfv1.MetricPipelineLabel, dfv1.MetricVertexLabel, dfv1.MetricPeriodLabel})
 
-	// Always expose metrics of following lookback seconds (1m, 5m, 15m)
+	// fixedLookbackSeconds Always expose metrics of following lookback seconds (1m, 5m, 15m)
 	fixedLookbackSeconds = map[string]int64{"1m": 60, "5m": 300, "15m": 900}
 )
 
diff --git a/server/apis/interface.go b/server/apis/interface.go
index 74f3bd17..2a47154e 100644
--- a/server/apis/interface.go
+++ b/server/apis/interface.go
@@ -8,8 +8,9 @@ type Handler interface {
 	ListNamespaces(c *gin.Context)
 	ListInterStepBufferServices(c *gin.Context)
 	GetInterStepBufferService(c *gin.Context)
-	ListVerices(c *gin.Context)
+	ListVertices(c *gin.Context)
 	GetVertex(c *gin.Context)
+	GetVertexMetrics(c *gin.Context)
 	ListVertexPods(c *gin.Context)
 	PodLogs(c *gin.Context)
 	ListPodsMetrics(c *gin.Context)
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index 958249dd..d13e0ba0 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -30,6 +30,7 @@ type handler struct {
 	numaflowClient dfv1clients.NumaflowV1alpha1Interface
 }
 
+// NewHandler is used to provide a new instance of the handler type
 func NewHandler() (*handler, error) {
 	var restConfig *rest.Config
 	var err error
@@ -62,6 +63,7 @@ func NewHandler() (*handler, error) {
 	}, nil
 }
 
+// ListPipelines is used to provide all the numaflow pipelines in a given namespace
 func (h *handler) ListPipelines(c *gin.Context) {
 	plList, err := h.numaflowClient.Pipelines(c.Param("namespace")).List(context.Background(), metav1.ListOptions{})
 	if err != nil {
@@ -71,6 +73,7 @@ func (h *handler) ListPipelines(c *gin.Context) {
 	c.JSON(http.StatusOK, plList.Items)
 }
 
+// GetPipeline is used to provide the spec of a given numaflow pipeline
 func (h *handler) GetPipeline(c *gin.Context) {
 	pl, err := h.numaflowClient.Pipelines(c.Param("namespace")).Get(context.Background(), c.Param("pipeline"), metav1.GetOptions{})
 	if err != nil {
@@ -80,6 +83,7 @@ func (h *handler) GetPipeline(c *gin.Context) {
 	c.JSON(http.StatusOK, pl)
 }
 
+// ListNamespaces is used to provide all the namespaces that have numaflow pipelines running
 func (h *handler) ListNamespaces(c *gin.Context) {
 	l, err := h.numaflowClient.Pipelines("").List(context.Background(), metav1.ListOptions{})
 	if err != nil {
@@ -97,6 +101,7 @@ func (h *handler) ListNamespaces(c *gin.Context) {
 	c.JSON(http.StatusOK, namespaces)
 }
 
+// ListInterStepBufferServices is used to provide all the interstepbuffer services in a namespace
 func (h *handler) ListInterStepBufferServices(c *gin.Context) {
 	limit, _ := strconv.ParseInt(c.Query("limit"), 10, 64)
 	isbSvcs, err := h.numaflowClient.InterStepBufferServices(c.Param("namespace")).List(context.Background(), metav1.ListOptions{
@@ -110,6 +115,7 @@ func (h *handler) ListInterStepBufferServices(c *gin.Context) {
 	c.JSON(http.StatusOK, isbSvcs.Items)
 }
 
+// GetInterStepBufferService is used to provide the spec of the interstep buffer service
 func (h *handler) GetInterStepBufferService(c *gin.Context) {
 	isbsvc, err := h.numaflowClient.InterStepBufferServices(c.Param("namespace")).Get(context.Background(), c.Param("isbsvc"), metav1.GetOptions{})
 	if err != nil {
@@ -119,7 +125,8 @@ func (h *handler) GetInterStepBufferService(c *gin.Context) {
 	c.JSON(http.StatusOK, isbsvc)
 }
 
-func (h *handler) ListVerices(c *gin.Context) {
+// ListVertices is used to provide all the vertices of a pipeline
+func (h *handler) ListVertices(c *gin.Context) {
 	limit, _ := strconv.ParseInt(c.Query("limit"), 10, 64)
 	vertices, err := h.numaflowClient.Vertices(c.Param("namespace")).List(context.Background(), metav1.ListOptions{
 		LabelSelector: fmt.Sprintf("%s=%s", dfv1.KeyPipelineName, c.Param("pipeline")),
@@ -133,6 +140,7 @@ func (h *handler) ListVerices(c *gin.Context) {
 	c.JSON(http.StatusOK, vertices.Items)
 }
 
+// GetVertex is used to provide the vertex spec
 func (h *handler) GetVertex(c *gin.Context) {
 	vertices, err := h.numaflowClient.Vertices(c.Param("namespace")).List(context.Background(), metav1.ListOptions{
 		LabelSelector: fmt.Sprintf("%s=%s,%s=%s", dfv1.KeyPipelineName, c.Param("pipeline"), dfv1.KeyVertexName, c.Param("vertex")),
@@ -148,6 +156,7 @@ func (h *handler) GetVertex(c *gin.Context) {
 	c.JSON(http.StatusOK, vertices.Items[0])
 }
 
+// ListVertexPods is used to provide all the pods of a vertex
 func (h *handler) ListVertexPods(c *gin.Context) {
 	limit, _ := strconv.ParseInt(c.Query("limit"), 10, 64)
 	pods, err := h.kubeClient.CoreV1().Pods(c.Param("namespace")).List(context.Background(), metav1.ListOptions{
@@ -162,6 +171,7 @@ func (h *handler) ListVertexPods(c *gin.Context) {
 	c.JSON(http.StatusOK, pods.Items)
 }
 
+// ListPodsMetrics is used to provide a list of all metrics in all the pods
 func (h *handler) ListPodsMetrics(c *gin.Context) {
 	limit, _ := strconv.ParseInt(c.Query("limit"), 10, 64)
 	l, err := h.metricsClient.MetricsV1beta1().PodMetricses(c.Param("namespace")).List(context.Background(), metav1.ListOptions{
@@ -175,6 +185,7 @@ func (h *handler) ListPodsMetrics(c *gin.Context) {
 	c.JSON(http.StatusOK, l.Items)
 }
 
+// GetPodMetrics is used to provide the metrics like CPU/Memory utilization for a pod
 func (h *handler) GetPodMetrics(c *gin.Context) {
 	m, err := h.metricsClient.MetricsV1beta1().PodMetricses(c.Param("namespace")).Get(context.Background(), c.Param("pod"), metav1.GetOptions{})
 	if err != nil {
@@ -184,6 +195,7 @@ func (h *handler) GetPodMetrics(c *gin.Context) {
 	c.JSON(http.StatusOK, m)
 }
 
+// PodLogs is used to provide the logs of a given container in pod
 func (h *handler) PodLogs(c *gin.Context) {
 	var tailLines *int64
 	if v := c.Query("tailLines"); v != "" {
@@ -210,6 +222,7 @@ func (h *handler) PodLogs(c *gin.Context) {
 	}
 }
 
+// ListPipelineEdges is used to provide information about all the pipeline edges
 func (h *handler) ListPipelineEdges(c *gin.Context) {
 	ns := c.Param("namespace")
 	pipeline := c.Param("pipeline")
@@ -226,6 +239,7 @@ func (h *handler) ListPipelineEdges(c *gin.Context) {
 	c.JSON(http.StatusOK, l)
 }
 
+// GetPipelineEdge is used to provide information about a single pipeline edge
 func (h *handler) GetPipelineEdge(c *gin.Context) {
 	ns := c.Param("namespace")
 	pipeline := c.Param("pipeline")
@@ -243,6 +257,24 @@ func (h *handler) GetPipelineEdge(c *gin.Context) {
 	c.JSON(http.StatusOK, i)
 }
 
+// GetVertexMetrics is used to provide information about the vertex including processing Rate
+func (h *handler) GetVertexMetrics(c *gin.Context) {
+	ns := c.Param("namespace")
+	pipeline := c.Param("pipeline")
+	vertex := c.Param("vertex")
+	client, err := daemonclient.NewDaemonServiceClient(daemonSvcAddress(ns, pipeline))
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, err.Error())
+		return
+	}
+	l, err := client.GetVertexMetrics(context.Background(), ns, pipeline, vertex)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, err.Error())
+		return
+	}
+	c.JSON(http.StatusOK, l)
+}
+
 func daemonSvcAddress(ns, pipeline string) string {
 	return fmt.Sprintf("%s.%s.svc.cluster.local:%d", fmt.Sprintf("%s-daemon-svc", pipeline), ns, dfv1.DaemonServicePort)
 }
diff --git a/server/routes/routes.go b/server/routes/routes.go
index 9e6d6efc..bacf6c2f 100644
--- a/server/routes/routes.go
+++ b/server/routes/routes.go
@@ -25,7 +25,7 @@ func v1Routes(r gin.IRouter) {
 	r.GET("/namespaces/:namespace/isbsvcs", handler.ListInterStepBufferServices)
 	r.GET("/namespaces/:namespace/isbsvcs/:isbsvc", handler.GetInterStepBufferService)
 	r.GET("/namespaces/:namespace/pipelines/:pipeline", handler.GetPipeline)
-	r.GET("/namespaces/:namespace/pipelines/:pipeline/vertices", handler.ListVerices)
+	r.GET("/namespaces/:namespace/pipelines/:pipeline/vertices", handler.ListVertices)
 	r.GET("/namespaces/:namespace/pipelines/:pipeline/vertices/:vertex", handler.GetVertex)
 	r.GET("/namespaces/:namespace/pipelines/:pipeline/vertices/:vertex/pods", handler.ListVertexPods)
 	r.GET("/namespaces/:namespace/pods/:pod/log", handler.PodLogs)
@@ -33,4 +33,5 @@ func v1Routes(r gin.IRouter) {
 	r.GET("/metrics/namespaces/:namespace/pods/:pod", handler.GetPodMetrics)
 	r.GET("/namespaces/:namespace/pipelines/:pipeline/edges", handler.ListPipelineEdges)
 	r.GET("/namespaces/:namespace/pipelines/:pipeline/edges/:edge", handler.GetPipelineEdge)
+	r.GET("/namespaces/:namespace/pipelines/:pipeline/vertices/:vertex/metrics", handler.GetVertexMetrics)
 }


Test patch:
diff --git a/pkg/daemon/server/service/pipeline_metrics_query_service_test.go b/pkg/daemon/server/service/pipeline_metrics_query_service_test.go
new file mode 100644
index 00000000..9af36298
--- /dev/null
+++ b/pkg/daemon/server/service/pipeline_metrics_query_service_test.go
@@ -0,0 +1,157 @@
+package service
+
+import (
+	"bytes"
+	"context"
+	"io/ioutil"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/apis/proto/daemon"
+	"github.com/numaproj/numaflow/pkg/isbsvc"
+)
+
+type mockGetType func(url string) (*http.Response, error)
+
+type mockHttpClient struct {
+	MockGet mockGetType
+}
+
+func (m *mockHttpClient) Get(url string) (*http.Response, error) {
+	return m.MockGet(url)
+
+}
+
+type mockIsbSvcClient struct {
+}
+
+func (ms *mockIsbSvcClient) GetBufferInfo(ctx context.Context, buffer v1alpha1.Buffer) (*isbsvc.BufferInfo, error) {
+	return &isbsvc.BufferInfo{
+		Name:            buffer.Name,
+		PendingCount:    10,
+		AckPendingCount: 15,
+		TotalMessages:   20,
+	}, nil
+}
+
+func (ms *mockIsbSvcClient) CreateBuffers(ctx context.Context, buffers []v1alpha1.Buffer, opts ...isbsvc.BufferCreateOption) error {
+	return nil
+}
+
+func (ms *mockIsbSvcClient) DeleteBuffers(ctx context.Context, buffers []v1alpha1.Buffer) error {
+	return nil
+}
+
+func (ms *mockIsbSvcClient) ValidateBuffers(ctx context.Context, buffers []v1alpha1.Buffer) error {
+	return nil
+}
+
+func TestGetVertexMetrics(t *testing.T) {
+	pipelineName := "simple-pipeline"
+	pipeline := &v1alpha1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: pipelineName},
+	}
+	client, _ := isbsvc.NewISBJetStreamSvc(pipelineName)
+	pipelineMetricsQueryService := NewPipelineMetricsQueryService(client, pipeline)
+
+	metricsResponse := `# HELP vertex_processing_rate Message processing rate in the last period of seconds, tps. It represents the rate of a vertex instead of a pod.
+# TYPE vertex_processing_rate gauge
+vertex_processing_rate{period="15m",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
+vertex_processing_rate{period="1m",pipeline="simple-pipeline",vertex="cat"} 5.084745762711864
+vertex_processing_rate{period="5m",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
+vertex_processing_rate{period="default",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
+`
+	ioReader := ioutil.NopCloser(bytes.NewReader([]byte(metricsResponse)))
+
+	pipelineMetricsQueryService.httpClient = &mockHttpClient{
+		MockGet: func(url string) (*http.Response, error) {
+			return &http.Response{
+				StatusCode: 200,
+				Body:       ioReader,
+			}, nil
+		},
+	}
+
+	vertex := "cat"
+	namespace := "numaflow-system"
+
+	req := &daemon.GetVertexMetricsRequest{Vertex: &vertex, Namespace: &namespace}
+
+	resp, err := pipelineMetricsQueryService.GetVertexMetrics(context.Background(), req)
+	assert.NoError(t, err)
+
+	processingRates := make(map[string]float32)
+
+	processingRates["15m"] = 4.894737
+	processingRates["1m"] = 5.084746
+	processingRates["5m"] = 4.894737
+	processingRates["default"] = 4.894737
+	assert.Equal(t, resp.Vertex.GetProcessingRates(), processingRates)
+}
+
+func TestGetBuffer(t *testing.T) {
+
+	pipelineName := "simple-pipeline"
+	namespace := "numaflow-system"
+	edges := []v1alpha1.Edge{
+		{
+			From: "in",
+			To:   "cat",
+		},
+	}
+	pipeline := &v1alpha1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      pipelineName,
+			Namespace: namespace,
+		},
+		Spec: v1alpha1.PipelineSpec{Edges: edges},
+	}
+
+	ms := &mockIsbSvcClient{}
+	pipelineMetricsQueryService := NewPipelineMetricsQueryService(ms, pipeline)
+
+	bufferName := "numaflow-system-simple-pipeline-in-cat"
+
+	req := &daemon.GetBufferRequest{Pipeline: &pipelineName, Buffer: &bufferName}
+
+	resp, err := pipelineMetricsQueryService.GetBuffer(context.Background(), req)
+	assert.NoError(t, err)
+	assert.Equal(t, *resp.Buffer.BufferUsage, 0.0004)
+
+}
+
+func TestListBuffers(t *testing.T) {
+
+	pipelineName := "simple-pipeline"
+	namespace := "numaflow-system"
+	edges := []v1alpha1.Edge{
+		{
+			From: "in",
+			To:   "cat",
+		},
+		{
+			From: "cat",
+			To:   "out",
+		},
+	}
+	pipeline := &v1alpha1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      pipelineName,
+			Namespace: namespace,
+		},
+		Spec: v1alpha1.PipelineSpec{Edges: edges},
+	}
+
+	ms := &mockIsbSvcClient{}
+	pipelineMetricsQueryService := NewPipelineMetricsQueryService(ms, pipeline)
+
+	req := &daemon.ListBuffersRequest{Pipeline: &pipelineName}
+
+	resp, err := pipelineMetricsQueryService.ListBuffers(context.Background(), req)
+	assert.NoError(t, err)
+	assert.Equal(t, len(resp.Buffers), 2)
+}