Repository: numaproj/numaflow
Pull request #59 (2022-06-15 22:25:38)

feat: auto-scaling (part 1)
Signed-off-by: Derek Wang <whynowy@gmail.com>

This is the first part of the code changes for auto-scaling feature.

1. An `Ratable` interface is introduced in this PR, which will be implemented in the ISB reader and writer to measure the processing rate. For ISB reader, the `Rate()` returns the acknowledge rate, in ISB writer, it returns the writing rate.

2. An `Pending()` function is also introduced into `Reader` interface, which is intended for the ISB Reader and sources to implement.


Patch:
diff --git a/docs/metrics/METRICS.md b/docs/metrics/METRICS.md
new file mode 100644
index 00000000..0ac3ef9f
--- /dev/null
+++ b/docs/metrics/METRICS.md
@@ -0,0 +1,157 @@
+# Metrics
+
+NumaFlow provides the following prometheus metrics which we can use to monitor our pipeline and setup any alerts if needed.
+
+## Golden Signals
+These metrics in combination can be used to determine the overall health of your pipeline
+
+### Traffic
+These metrics can be used to determine throughput of your pipeline.
+
+#### Data-forward
+
+
+| Metric name             | Metric type | Labels                                                                                               | Description                                                                                    |
+|-------------------------|-------------|------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------|
+| `forwarder_read_total`  | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Provides the total number of messages read by a given Vertex from an Inter-Step Buffer         |
+| `forwarder_write_total` | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Provides the total number of messages written to Inter-Step Buffer by a given Vertex           |
+| `forwarder_ack_total`   | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Provides the total number of messages acknowledged by a given Vertex from an Inter-Step Buffer |
+
+#### Kafka Source
+
+| Metric name               | Metric type | Labels                                                               | Description                                                                       |
+|---------------------------|-------------|----------------------------------------------------------------------|-----------------------------------------------------------------------------------|
+| `kafka_source_read_total` | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;   | Provides the number of messages read by the Kafka Source Vertex/Processor.        |
+| `kafka_source_ack_total`  | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;   | Provides the number of messages acknowledged by the Kafka Source Vertex/Processor |
+
+
+#### Generator Source
+
+| Metric name                 | Metric type | Labels                                                             | Description                                                                    |
+|-----------------------------|-------------|--------------------------------------------------------------------|--------------------------------------------------------------------------------|
+| `tickgen_source_read_total` | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; | Provides the number of messages read by the Generator Source Vertex/Processor. |
+
+#### Http Source
+
+| Metric name              | Metric type | Labels                                                              | Description                                                               |
+|--------------------------|-------------|---------------------------------------------------------------------|---------------------------------------------------------------------------|
+| `http_source_read_total` | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;  | Provides the number of messages read by the HTTP Source Vertex/Processor. |
+
+
+#### Kafka Sink
+
+| Metric name              | Metric type | Labels                                                              | Description                                                                |
+|--------------------------|-------------|---------------------------------------------------------------------|----------------------------------------------------------------------------|
+| `kafka_sink_write_total` | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;  | Provides the number of messages written by the Kafka Sink Vertex/Processor |
+
+
+#### Log Sink
+
+| Metric name            | Metric type | Labels                                                              | Description                                                              |
+|------------------------|-------------|---------------------------------------------------------------------|--------------------------------------------------------------------------|
+| `log_sink_write_total` | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;  | Provides the number of messages written by the Log Sink Vertex/Processor |
+
+### Latency 
+These metrics can be used to determine the latency of your pipeline.
+
+| Metric name                               | Metric type | Labels                                                                                               | Description                                                                                    |
+|-------------------------------------------|-------------|------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------|
+| `forwarder_udf_processing_time`           | Histogram   | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Provides a histogram distribution of the processing times of User Defined Functions. (UDF's)   |
+| `forwarder_forward_chunk_processing_time` | Histogram   | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Provides a histogram distribution of the processing times of the forwarder function as a whole |
+
+
+### Errors
+These metrics can be used to determine if there are any errors in the pipeline
+
+
+| Metric name                       | Metric type | Labels                                                                                               | Description                                                        |
+|-----------------------------------|-------------|------------------------------------------------------------------------------------------------------|--------------------------------------------------------------------|
+| `forwarder_platform_error`        | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Indicates any internal errors which could stop pipeline processing |
+| `forwarder_read_error`            | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Indicates any errors while reading messages by the forwarder       |
+| `forwarder_write_error`           | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Indicates any errors while writing messages by the forwarder       |
+| `forwarder_ack_error`             | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt; <br> `buffer`=&lt;buffer-name&gt; | Indicates any errors while acknowledging messages by the forwarder |
+| `kafka_source_offset_ack_errors`  | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;                                   | Indicates any kafka acknowledgement errors                         |
+| `kafka_sink_write_error_total`    | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;                                   | Provides the number of errors while writing to the Kafka sink      |
+| `kafka_sink_write_timeout_total`  | Counter     | `vertex`=&lt;vertex-name&gt; <br> `pipeline`=&lt;pipeline-name&gt;                                   | Provides the write timeouts  while writing to the Kafka sink       |
+| `isb_jetstream_read_error_total`  | Counter     | `buffer`=&lt;buffer-name&gt;                                                                         | Indicates any read errors with NATS Jetstream ISB                  |
+| `isb_jetstream_write_error_total` | Counter     | `buffer`=&lt;buffer-name&gt;                                                                         | Indicates any write errors  with NATS Jetstream ISB                |
+| `isb_redis_read_error_total`      | Counter     | `buffer`=&lt;buffer-name&gt;                                                                         | Indicates any read errors with Redis ISB                           |
+| `isb_redis_write_error_total`     | Counter     | `buffer`=&lt;buffer-name&gt;                                                                         | Indicates any write errors with Redis ISB                          |
+
+### Saturation
+
+#### NATS JetStream ISB
+
+| Metric name                        | Metric type | Labels                       | Description                                                                                                                                  |
+|------------------------------------|-------------|------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------|
+| `isb_jetstream_isFull_total`       | Counter     | `buffer`=&lt;buffer-name&gt; | Indicates if the ISB is full. Continual increase of this counter metric indicates a potential backpressure that can be built on the pipeline |
+| `isb_jetstream_buffer_soft_usage`  | Gauge       | `buffer`=&lt;buffer-name&gt; | Indicates the usage/utilization of a NATS Jetstream ISB                                                                                      |
+| `isb_jetstream_buffer_solid_usage` | Gauge       | `buffer`=&lt;buffer-name&gt; | Indicates the solid usage of a NATS Jetstream ISB                                                                                            |
+| `isb_jetstream_buffer_pending`     | Gauge       | `buffer`=&lt;buffer-name&gt; | Indicate the number of pending messages at a given point in time.                                                                            |
+| `isb_jetstream_buffer_ack_pending` | Gauge       | `buffer`=&lt;buffer-name&gt; | Indicates the number of messages pending acknowledge at a given point in time                                                                |
+
+
+#### Redis ISB
+
+| Metric name              | Metric type | Labels                       | Description                                                                                                                                  |
+|--------------------------|-------------|------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------|
+| `isb_redis_isFull_total` | Counter     | `buffer`=&lt;buffer-name&gt; | Indicates if the ISB is full. Continual increase of this counter metric indicates a potential backpressure that can be built on the pipeline |
+| `isb_redis_buffer_usage` | Gauge       | `buffer`=&lt;buffer-name&gt; | Indicates the usage/utilization of a Redis ISB                                                                                               |
+| `isb_redis_consumer_lag` | Gauge       | `buffer`=&lt;buffer-name&gt; | Indicates the the consumer lag of a Redis ISB                                                                                                |
+
+
+## Prometheus Operator for Scraping Metrics:
+
+You can follow the [prometheus operator](https://github.com/prometheus-operator/prometheus-operator/blob/main/Documentation/user-guides/getting-started.md) setup guide if you would like to use prometheus operator configured in your cluster.
+
+### Configure the below Service Monitors for scraping your pipeline metrics:
+```yaml
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  labels:
+    app.kubernetes.io/part-of: numaflow
+  name: numaflow-pipeline-metrics
+spec:
+  endpoints:
+    - scheme: https
+      port: metrics
+      targetPort: 2469
+      tlsConfig:
+        insecureSkipVerify: true
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: vertex
+      app.kubernetes.io/managed-by: vertex-controller
+      app.kubernetes.io/part-of: numaflow
+    matchExpressions:
+    - key: numaflow.numaproj.io/pipeline-name
+      operator: Exists
+    - key: numaflow.numaproj.io/vertex-name
+      operator: Exists
+```
+
+### Configure the below Service Monitor if you use the NATS Jetstream ISB for your NATS Jetstream metrics:
+
+```yaml
+apiVersion: monitoring.coreos.com/v1
+kind: ServiceMonitor
+metadata:
+  labels:
+    app.kubernetes.io/part-of: numaflow
+  name: numaflow-isbsvc-jetstream-metrics
+spec:
+  endpoints:
+    - scheme: http
+      port: metrics
+      targetPort: 7777
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: isbsvc
+      app.kubernetes.io/managed-by: isbsvc-controller
+      app.kubernetes.io/part-of: numaflow
+      numaflow.numaproj.io/isbsvc-type: jetstream
+    matchExpressions:
+    - key: numaflow.numaproj.io/isbsvc-name
+      operator: Exists
+```
diff --git a/pkg/daemon/server/daemon_server.go b/pkg/daemon/server/daemon_server.go
index fd329e02..75ed5124 100644
--- a/pkg/daemon/server/daemon_server.go
+++ b/pkg/daemon/server/daemon_server.go
@@ -62,7 +62,7 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 			return err
 		}
 	default:
-		return fmt.Errorf("unsupported isbs buffer type %q", ds.isbSvcType)
+		return fmt.Errorf("unsupported isbsvc buffer type %q", ds.isbSvcType)
 	}
 
 	cer, err := sharedtls.GenerateX509KeyPair()
diff --git a/pkg/isb/interfaces.go b/pkg/isb/interfaces.go
index e66fa851..1b2f7dca 100644
--- a/pkg/isb/interfaces.go
+++ b/pkg/isb/interfaces.go
@@ -12,6 +12,17 @@ import (
 	"strconv"
 )
 
+const (
+	PendingNotAvailable = int64(-1)
+	RateNotAvailable    = float64(-1)
+)
+
+// Ratable is the interface that wraps the Rate method.
+type Ratable interface {
+	// Rate returns the rough rate (messages/second), this is used for auto-scaling calculation
+	Rate(context.Context) (float64, error)
+}
+
 // BufferWriter is the buffer to which we are writing.
 type BufferWriter interface {
 	BufferWriterInformation
@@ -31,6 +42,8 @@ type BufferReader interface {
 	Read(context.Context, int64) ([]*ReadMessage, error)
 	// Ack acknowledges an array of offset.
 	Ack(context.Context, []Offset) []error
+	// Pending returns the pending messages number.
+	Pending(context.Context) (int64, error)
 }
 
 // BufferReaderInformation has information regarding the buffer we are reading from.
diff --git a/pkg/isb/jetstream/options.go b/pkg/isb/jetstream/options.go
index edf2ccce..04c4b6e0 100644
--- a/pkg/isb/jetstream/options.go
+++ b/pkg/isb/jetstream/options.go
@@ -14,13 +14,19 @@ type writeOptions struct {
 	bufferUsageLimit float64
 	// refreshInterval is used to provide the default refresh interval
 	refreshInterval time.Duration
+	// useWriteInfoAsRate indicates whether to check the write sequence for rate calculation
+	useWriteInfoAsRate bool
+	// rateLookbackSeconds is the look back seconds for rate calculation
+	rateLookbackSeconds int64
 }
 
 func defaultWriteOptions() *writeOptions {
 	return &writeOptions{
-		maxLength:        dfv1.DefaultBufferLength,
-		bufferUsageLimit: dfv1.DefaultBufferUsageLimit,
-		refreshInterval:  1 * time.Second,
+		maxLength:           dfv1.DefaultBufferLength,
+		bufferUsageLimit:    dfv1.DefaultBufferUsageLimit,
+		refreshInterval:     1 * time.Second,
+		useWriteInfoAsRate:  false,
+		rateLookbackSeconds: 180,
 	}
 }
 
@@ -50,10 +56,24 @@ func WithRefreshInterval(refreshInterval time.Duration) WriteOption {
 	}
 }
 
+// WithUsingWriteInfoAsRate sets whether to check sequence for rate calculation
+func WithUsingWriteInfoAsRate(check bool) WriteOption {
+	return func(o *writeOptions) error {
+		o.useWriteInfoAsRate = check
+		return nil
+	}
+}
+
 // options for reading from JetStream
 type readOptions struct {
 	// readTimeOut is the timeout needed for read timeout
 	readTimeOut time.Duration
+	// Whether to run ack information check
+	useAckInfoAsRate bool
+	// ackCheckInterval is the interval for stream information check such as pending messages
+	ackInfoCheckInterval time.Duration
+	// rateLookbackSeconds is the look back seconds for rate calculation
+	rateLookbackSeconds int64
 }
 
 type ReadOption func(*readOptions) error
@@ -65,8 +85,35 @@ func WithReadTimeOut(timeout time.Duration) ReadOption {
 		return nil
 	}
 }
+
+// WithUsingAckInfoAsRate is used to set whether to run ack information check in the reader, which is for ack rate calculation
+func WithUsingAckInfoAsRate(yes bool) ReadOption {
+	return func(o *readOptions) error {
+		o.useAckInfoAsRate = yes
+		return nil
+	}
+}
+
+// WithAckInfoCheckInterval is used to set the periodical ack information check interval
+func WithAckInfoCheckInterval(t time.Duration) ReadOption {
+	return func(o *readOptions) error {
+		o.ackInfoCheckInterval = t
+		return nil
+	}
+}
+
+func WithRateLookbackSeconds(seconds int64) ReadOption {
+	return func(o *readOptions) error {
+		o.rateLookbackSeconds = seconds
+		return nil
+	}
+}
+
 func defaultReadOptions() *readOptions {
 	return &readOptions{
-		readTimeOut: time.Second,
+		readTimeOut:          time.Second,
+		useAckInfoAsRate:     false,
+		ackInfoCheckInterval: 3 * time.Second,
+		rateLookbackSeconds:  180,
 	}
 }
diff --git a/pkg/isb/jetstream/reader.go b/pkg/isb/jetstream/reader.go
index edcccf14..8933d744 100644
--- a/pkg/isb/jetstream/reader.go
+++ b/pkg/isb/jetstream/reader.go
@@ -14,6 +14,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isbsvc/clients"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
+	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
 )
 
 type jetStreamReader struct {
@@ -21,10 +22,13 @@ type jetStreamReader struct {
 	stream                string
 	subject               string
 	conn                  *nats.Conn
+	js                    nats.JetStreamContext
 	sub                   *nats.Subscription
 	opts                  *readOptions
 	inProgessTickDuration time.Duration
 	log                   *zap.SugaredLogger
+	// ackedInfo stores a list of ack seq/timestamp(seconds) information
+	ackedInfo *sharedqueue.OverflowQueue[timestampedSequence]
 }
 
 // NewJetStreamBufferReader is used to provide a new JetStream buffer reader connection
@@ -76,11 +80,16 @@ func NewJetStreamBufferReader(ctx context.Context, client clients.JetStreamClien
 		stream:                stream,
 		subject:               subject,
 		conn:                  conn,
+		js:                    js,
 		sub:                   sub,
 		opts:                  o,
 		inProgessTickDuration: time.Duration(inProgessTickSeconds * int64(time.Second)),
 		log:                   log,
 	}
+	if o.useAckInfoAsRate {
+		result.ackedInfo = sharedqueue.New[timestampedSequence](1800)
+		go result.runAckInfomationChecker(ctx)
+	}
 	return result, nil
 }
 
@@ -100,6 +109,68 @@ func (jr *jetStreamReader) Close() error {
 	return nil
 }
 
+func (jr *jetStreamReader) runAckInfomationChecker(ctx context.Context) {
+	js, err := jr.conn.JetStream()
+	if err != nil {
+		// Let it exit if it fails to start the information checker
+		jr.log.Fatal("Failed to get Jet Stream context, %w", err)
+	}
+	checkAckInfo := func() {
+		c, err := js.ConsumerInfo(jr.stream, jr.stream)
+		if err != nil {
+			jr.log.Errorw("failed to get consumer info in the reader", zap.Error(err))
+			return
+		}
+		ts := timestampedSequence{seq: int64(c.AckFloor.Stream), timestamp: time.Now().Unix()}
+		jr.ackedInfo.Append(ts)
+		jr.log.Debugw("Acknowledge information", zap.Int64("ackSeq", ts.seq), zap.Int64("timestamp", ts.timestamp))
+	}
+	checkAckInfo()
+
+	ticker := time.NewTicker(jr.opts.ackInfoCheckInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			checkAckInfo()
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func (jr *jetStreamReader) Pending(_ context.Context) (int64, error) {
+	c, err := jr.js.ConsumerInfo(jr.stream, jr.stream)
+	if err != nil {
+		return isb.PendingNotAvailable, fmt.Errorf("failed to get consumer info, %w", err)
+	}
+	return int64(c.NumPending), nil
+}
+
+// Rate returns the ack rate (tps)
+func (jr *jetStreamReader) Rate(_ context.Context) (float64, error) {
+	if !jr.opts.useAckInfoAsRate {
+		return isb.RateNotAvailable, nil
+	}
+	timestampedSeqs := jr.ackedInfo.Items()
+	if len(timestampedSeqs) < 2 {
+		return isb.RateNotAvailable, nil
+	}
+	endSeqInfo := timestampedSeqs[len(timestampedSeqs)-1]
+	startSeqInfo := timestampedSeqs[len(timestampedSeqs)-2]
+	for i := len(timestampedSeqs) - 3; i >= 0; i-- {
+		if endSeqInfo.timestamp-timestampedSeqs[i].timestamp > jr.opts.rateLookbackSeconds {
+			startSeqInfo = timestampedSeqs[i]
+			break
+		}
+	}
+	// Check if it is too stale (use lookbackSeconds + 2 * interval to determine)
+	if endSeqInfo.timestamp-startSeqInfo.timestamp > jr.opts.rateLookbackSeconds+2*int64(jr.opts.ackInfoCheckInterval.Seconds()) {
+		return isb.RateNotAvailable, nil
+	}
+	return float64(endSeqInfo.seq-startSeqInfo.seq) / float64(endSeqInfo.timestamp-startSeqInfo.timestamp), nil
+}
+
 func (jr *jetStreamReader) Read(_ context.Context, count int64) ([]*isb.ReadMessage, error) {
 	result := []*isb.ReadMessage{}
 	msgs, err := jr.sub.Fetch(int(count), nats.MaxWait(jr.opts.readTimeOut))
diff --git a/pkg/isb/jetstream/types.go b/pkg/isb/jetstream/types.go
new file mode 100644
index 00000000..ac106e3c
--- /dev/null
+++ b/pkg/isb/jetstream/types.go
@@ -0,0 +1,8 @@
+package jetstream
+
+// TimestampedSequence is a helper struct to wrap a sequence ID and timestamp pair
+type timestampedSequence struct {
+	seq int64
+	// timestamp in seconds
+	timestamp int64
+}
diff --git a/pkg/isb/jetstream/writer.go b/pkg/isb/jetstream/writer.go
index 934eec45..6f3fed3c 100644
--- a/pkg/isb/jetstream/writer.go
+++ b/pkg/isb/jetstream/writer.go
@@ -13,6 +13,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isbsvc/clients"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
+	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
 )
 
 type jetStreamWriter struct {
@@ -25,6 +26,8 @@ type jetStreamWriter struct {
 	log     *zap.SugaredLogger
 
 	isFull *atomic.Bool
+	// writtenInfo stores a list of written seq/timestamp(seconds) information
+	writtenInfo *sharedqueue.OverflowQueue[timestampedSequence]
 }
 
 // NewJetStreamBufferWriter is used to provide a new instance of JetStreamBufferWriter
@@ -59,6 +62,10 @@ func NewJetStreamBufferWriter(ctx context.Context, client clients.JetStreamClien
 		log:     logging.FromContext(ctx).With("bufferWriter", name).With("stream", stream).With("subject", subject),
 	}
 
+	if o.useWriteInfoAsRate {
+		result.writtenInfo = sharedqueue.New[timestampedSequence](3600)
+	}
+
 	go result.runStatusChecker(ctx)
 	return result, nil
 }
@@ -75,13 +82,18 @@ func (jw *jetStreamWriter) runStatusChecker(ctx context.Context) {
 		s, err := js.StreamInfo(jw.stream)
 		if err != nil {
 			isbFullErrors.With(labels).Inc()
-			jw.log.Errorw("Failed to get stream info", zap.Error(err))
+			jw.log.Errorw("Failed to get stream info in the writer", zap.Error(err))
 			return
 		}
+		if jw.opts.useWriteInfoAsRate {
+			ts := timestampedSequence{seq: int64(s.State.LastSeq), timestamp: time.Now().Unix()}
+			jw.writtenInfo.Append(ts)
+			jw.log.Debugw("Written information", zap.Int64("lastSeq", ts.seq), zap.Int64("timestamp", ts.timestamp))
+		}
 		c, err := js.ConsumerInfo(jw.stream, jw.stream)
 		if err != nil {
 			isbFullErrors.With(labels).Inc()
-			jw.log.Errorw("failed to get consumer info", zap.Error(err))
+			jw.log.Errorw("failed to get consumer info in the writer", zap.Error(err))
 			return
 		}
 		var solidUsage, softUsage float64
@@ -134,6 +146,30 @@ func (jw *jetStreamWriter) Close() error {
 	return nil
 }
 
+// Rate returns the writting rate (tps)
+func (jw *jetStreamWriter) Rate(_ context.Context) (float64, error) {
+	if jw.opts.useWriteInfoAsRate {
+		return isb.RateNotAvailable, nil
+	}
+	timestampedSeqs := jw.writtenInfo.Items()
+	if len(timestampedSeqs) < 2 {
+		return isb.RateNotAvailable, nil
+	}
+	endSeqInfo := timestampedSeqs[len(timestampedSeqs)-1]
+	startSeqInfo := timestampedSeqs[len(timestampedSeqs)-1]
+	for i := len(timestampedSeqs) - 3; i >= 0; i-- {
+		if endSeqInfo.timestamp-timestampedSeqs[i].timestamp > jw.opts.rateLookbackSeconds {
+			startSeqInfo = timestampedSeqs[i]
+			break
+		}
+	}
+	// Check if it is too stale (use lookbackSeconds + 4 * interval to determine)
+	if endSeqInfo.timestamp-startSeqInfo.timestamp > jw.opts.rateLookbackSeconds+4*int64(jw.opts.refreshInterval.Seconds()) {
+		return isb.RateNotAvailable, nil
+	}
+	return float64(endSeqInfo.seq-startSeqInfo.seq) / float64(endSeqInfo.timestamp-startSeqInfo.timestamp), nil
+}
+
 func (jw *jetStreamWriter) Write(_ context.Context, messages []isb.Message) ([]isb.Offset, []error) {
 	labels := map[string]string{"buffer": jw.GetName()}
 	var errs = make([]error, len(messages))
diff --git a/pkg/isb/redis/read.go b/pkg/isb/redis/read.go
index 8c09481e..cf8ac38a 100644
--- a/pkg/isb/redis/read.go
+++ b/pkg/isb/redis/read.go
@@ -310,6 +310,16 @@ func (br *BufferRead) convertXStreamToMessages(xstreams []redis.XStream, message
 	return messages, nil
 }
 
+func (br *BufferRead) Pending(_ context.Context) (int64, error) {
+	// TODO: not implemented
+	return isb.PendingNotAvailable, nil
+}
+
+func (br *BufferRead) Rate(_ context.Context) (float64, error) {
+	// TODO: not implemented
+	return isb.RateNotAvailable, nil
+}
+
 func getHeaderAndBody(field string, value interface{}) (msg isb.Message, err error) {
 	err = msg.Header.UnmarshalBinary([]byte(field))
 	if err != nil {
diff --git a/pkg/isb/simplebuffer/buffer.go b/pkg/isb/simplebuffer/buffer.go
index adec8e94..08076b45 100644
--- a/pkg/isb/simplebuffer/buffer.go
+++ b/pkg/isb/simplebuffer/buffer.go
@@ -64,6 +64,16 @@ func (b *InMemoryBuffer) GetName() string {
 	return b.name
 }
 
+func (b *InMemoryBuffer) Pending(_ context.Context) (int64, error) {
+	// TODO: not implemented
+	return isb.PendingNotAvailable, nil
+}
+
+func (b *InMemoryBuffer) Rate(_ context.Context) (float64, error) {
+	// TODO: not implemented
+	return isb.RateNotAvailable, nil
+}
+
 // Close does nothing.
 func (b *InMemoryBuffer) Close() error {
 	return nil
diff --git a/pkg/shared/queue/overflow_queue.go b/pkg/shared/queue/overflow_queue.go
new file mode 100644
index 00000000..5d099ed6
--- /dev/null
+++ b/pkg/shared/queue/overflow_queue.go
@@ -0,0 +1,56 @@
+package queue
+
+import "sync"
+
+// OverflowQeueue is a thread safe queue implementation with max size, and the oldest elements automatically overflow.
+type OverflowQueue[T any] struct {
+	elements []T
+	maxSize  int
+	lock     *sync.RWMutex
+}
+
+func New[T any](size int) *OverflowQueue[T] {
+	return &OverflowQueue[T]{
+		elements: []T{},
+		maxSize:  size,
+		lock:     new(sync.RWMutex),
+	}
+}
+
+// Append adds an element to the queue
+func (q *OverflowQueue[T]) Append(value T) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	if len(q.elements) >= q.maxSize {
+		q.elements = q.elements[1:]
+	}
+	q.elements = append(q.elements, value)
+}
+
+// Items returns a copy of the elements in the queue
+func (q *OverflowQueue[T]) Items() []T {
+	q.lock.RLock()
+	defer q.lock.RUnlock()
+	r := make([]T, len(q.elements))
+	_ = copy(r, q.elements)
+	return r
+}
+
+// ReversedItems returns a reversed copy of the elements in the queue
+func (q *OverflowQueue[T]) ReversedItems() []T {
+	return reverse(q.Items())
+}
+
+// Length returns the current length of the queue
+func (q *OverflowQueue[T]) Length() int {
+	q.lock.RLock()
+	defer q.lock.RUnlock()
+	return len(q.elements)
+}
+
+func reverse[T any](input []T) []T {
+	if len(input) == 0 {
+		return input
+	}
+	return append(reverse(input[1:]), input[0])
+}
diff --git a/pkg/sinks/logger/log.go b/pkg/sinks/logger/log.go
index 441d1b4f..1d0789e3 100644
--- a/pkg/sinks/logger/log.go
+++ b/pkg/sinks/logger/log.go
@@ -6,11 +6,12 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 
+	"go.uber.org/zap"
+
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
-	"go.uber.org/zap"
 )
 
 // ToLog prints the output to a log sinks.
@@ -76,7 +77,7 @@ func (s *ToLog) IsFull() bool {
 func (s *ToLog) Write(_ context.Context, messages []isb.Message) ([]isb.Offset, []error) {
 	prefix := "(" + s.GetName() + ")"
 	for _, message := range messages {
-		logSinkReadCount.With(map[string]string{"vertex": s.name, "pipeline": s.pipelineName}).Inc()
+		logSinkWriteCount.With(map[string]string{"vertex": s.name, "pipeline": s.pipelineName}).Inc()
 		log.Println(prefix, string(message.Payload))
 	}
 	return nil, make([]error, len(messages))
diff --git a/pkg/sinks/logger/metrics.go b/pkg/sinks/logger/metrics.go
index a07d7e43..b99f429f 100644
--- a/pkg/sinks/logger/metrics.go
+++ b/pkg/sinks/logger/metrics.go
@@ -5,8 +5,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-// logSinkErrors is used to indicate the number of messages written to log sink
-var logSinkReadCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// logSinkWriteCount is used to indicate the number of messages written to log sink
+var logSinkWriteCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "log_sink",
 	Name:      "write_total",
 	Help:      "Total number of messages written to log sink",
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index f3843773..a5e05b84 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -252,6 +252,10 @@ func (mg *memgen) Start() <-chan struct{} {
 	return mg.forwarder.Start()
 }
 
+func (mg *memgen) Pending(_ context.Context) (int64, error) {
+	return isb.PendingNotAvailable, nil
+}
+
 // generator fires once per time unit and generates records and writes them to the channel
 func (mg *memgen) generator(ctx context.Context, rate int, timeunit time.Duration) {
 	go func() {
diff --git a/pkg/sources/http/http.go b/pkg/sources/http/http.go
index 635059c3..59639f12 100644
--- a/pkg/sources/http/http.go
+++ b/pkg/sources/http/http.go
@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"go.uber.org/zap"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
 	"github.com/numaproj/numaflow/pkg/isb/forward"
@@ -16,16 +18,16 @@ import (
 	sharedtls "github.com/numaproj/numaflow/pkg/shared/tls"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
-	"go.uber.org/zap"
 )
 
 type httpSource struct {
-	name        string
-	ready       bool
-	readTimeout time.Duration
-	bufferSize  int
-	messages    chan *isb.ReadMessage
-	logger      *zap.SugaredLogger
+	name         string
+	pipelineName string
+	ready        bool
+	readTimeout  time.Duration
+	bufferSize   int
+	messages     chan *isb.ReadMessage
+	logger       *zap.SugaredLogger
 
 	forwarder *forward.InterStepDataForward
 	shutdown  func(context.Context) error
@@ -58,10 +60,11 @@ func WithBufferSize(s int) Option {
 
 func New(vertex *dfv1.Vertex, writers []isb.BufferWriter, opts ...Option) (*httpSource, error) {
 	h := &httpSource{
-		name:        vertex.Spec.Name,
-		ready:       false,
-		bufferSize:  1000,            // default size
-		readTimeout: 1 * time.Second, // default timeout
+		name:         vertex.Spec.Name,
+		pipelineName: vertex.Spec.PipelineName,
+		ready:        false,
+		bufferSize:   1000,            // default size
+		readTimeout:  1 * time.Second, // default timeout
 	}
 	for _, o := range opts {
 		operr := o(h)
@@ -171,6 +174,10 @@ func (h *httpSource) GetName() string {
 	return h.name
 }
 
+func (h *httpSource) Pending(_ context.Context) (int64, error) {
+	return isb.PendingNotAvailable, nil
+}
+
 func (h *httpSource) Read(ctx context.Context, count int64) ([]*isb.ReadMessage, error) {
 	msgs := []*isb.ReadMessage{}
 	timeout := time.After(h.readTimeout)
@@ -183,6 +190,7 @@ func (h *httpSource) Read(ctx context.Context, count int64) ([]*isb.ReadMessage,
 			return msgs, nil
 		}
 	}
+	httpSourceReadCount.With(map[string]string{"vertex": h.name, "pipeline": h.pipelineName}).Inc()
 	h.logger.Debugf("Read %d messages.", len(msgs))
 	return msgs, nil
 }
diff --git a/pkg/sources/http/metrics.go b/pkg/sources/http/metrics.go
new file mode 100644
index 00000000..f90fd49e
--- /dev/null
+++ b/pkg/sources/http/metrics.go
@@ -0,0 +1,13 @@
+package http
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+// httpSourceReadCount is used to indicate the number of messages read by the http source vertex
+var httpSourceReadCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "http_source",
+	Name:      "read_total",
+	Help:      "Total number of messages Read",
+}, []string{"vertex", "pipeline"})
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index 0edc587a..0b6aeee0 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -165,6 +165,11 @@ func (r *KafkaSource) Close() error {
 	return nil
 }
 
+func (r *KafkaSource) Pending(ctx context.Context) (int64, error) {
+	// TODO: not implemented
+	return isb.PendingNotAvailable, nil
+}
+
 // NewKafkaSource returns a KafkaSource reader based on Kafka Consumer Group .
 func NewKafkaSource(vertex *dfv1.Vertex, writers []isb.BufferWriter, opts ...Option) (*KafkaSource, error) {
 	source := vertex.Spec.Source.Kafka


Test patch:
diff --git a/pkg/shared/queue/overflow_queue_test.go b/pkg/shared/queue/overflow_queue_test.go
new file mode 100644
index 00000000..6f69962c
--- /dev/null
+++ b/pkg/shared/queue/overflow_queue_test.go
@@ -0,0 +1,36 @@
+package queue
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAppend(t *testing.T) {
+	q := New[int](2)
+	q.Append(1)
+	q.Append(2)
+	assert.Equal(t, 2, q.Length())
+	assert.Contains(t, q.Items(), 1)
+	assert.Contains(t, q.Items(), 2)
+	q.Append(3)
+	assert.Equal(t, 2, q.Length())
+	assert.Contains(t, q.Items(), 2)
+	assert.Contains(t, q.Items(), 3)
+	q.Append(4)
+	q.Append(5)
+	q.Append(6)
+	assert.Equal(t, 5, q.Items()[0])
+	assert.Equal(t, 6, q.Items()[1])
+	assert.Equal(t, 6, q.ReversedItems()[0])
+	assert.Equal(t, 5, q.ReversedItems()[1])
+}
+
+func TestReverse(t *testing.T) {
+	l := []int{1, 2, 3}
+	l1 := reverse(l)
+	assert.Equal(t, 3, len(l1))
+	assert.Equal(t, 3, l1[0])
+	assert.Equal(t, 2, l1[1])
+	assert.Equal(t, 1, l1[2])
+}