feat: expose pending messages and processing rate (#79)

* feat: expose pending messages and processing rate Signed-off-by: Derek Wang <[email protected]>
numaproj · Jul 8, 2022 · 5592bb1 · 5592bb1
1 parent ac1814d
commit 5592bb1
Show file tree

Hide file tree

Showing 28 changed files with 986 additions and 413 deletions.
diff --git a/cmd/commands/processor.go b/cmd/commands/processor.go
@@ -52,30 +52,29 @@ func NewProcessorCommand() *cobra.Command {
 				return fmt.Errorf("invalid replica %q", replicaStr)
 			}
 			log = log.With("vertex", vertex.Name)
+			vertexInstance := &dfv1.VertexInstance{
+				Vertex:   vertex,
+				Hostname: hostname,
+				Replica:  int32(replica),
+			}
 			ctx := logging.WithLogger(signals.SetupSignalHandler(), log)
 			switch processorType {
 			case "source":
 				p := &sources.SourceProcessor{
-					ISBSvcType: dfv1.ISBSvcType(isbSvcType),
-					Vertex:     vertex,
-					Hostname:   hostname,
-					Replica:    replica,
+					ISBSvcType:     dfv1.ISBSvcType(isbSvcType),
+					VertexInstance: vertexInstance,
 				}
 				return p.Start(ctx)
 			case "sink":
 				p := &sinks.SinkProcessor{
-					ISBSvcType: dfv1.ISBSvcType(isbSvcType),
-					Vertex:     vertex,
-					Hostname:   hostname,
-					Replica:    replica,
+					ISBSvcType:     dfv1.ISBSvcType(isbSvcType),
+					VertexInstance: vertexInstance,
 				}
 				return p.Start(ctx)
 			case "udf":
 				p := &udf.UDFProcessor{
-					ISBSvcType: dfv1.ISBSvcType(isbSvcType),
-					Vertex:     vertex,
-					Hostname:   hostname,
-					Replica:    replica,
+					ISBSvcType:     dfv1.ISBSvcType(isbSvcType),
+					VertexInstance: vertexInstance,
 				}
 				return p.Start(ctx)
 			default:

diff --git a/config/base/crds/numaflow.numaproj.io_pipelines.yaml b/config/base/crds/numaflow.numaproj.io_pipelines.yaml
@@ -1444,8 +1444,14 @@ spec:
                       type: string
                     scale:
                       properties:
+                        lookbackSeconds:
+                          default: 180
+                          description: Lookback seconds to calculate the average pending
+                            messages and processing rate
+                          format: int32
+                          type: integer
                         max:
-                          default: 1
+                          default: 100
                           description: Maximum replicas
                           format: int32
                           type: integer

diff --git a/config/base/crds/numaflow.numaproj.io_vertices.yaml b/config/base/crds/numaflow.numaproj.io_vertices.yaml
@@ -1316,8 +1316,14 @@ spec:
                 type: integer
               scale:
                 properties:
+                  lookbackSeconds:
+                    default: 180
+                    description: Lookback seconds to calculate the average pending
+                      messages and processing rate
+                    format: int32
+                    type: integer
                   max:
-                    default: 1
+                    default: 100
                     description: Maximum replicas
                     format: int32
                     type: integer

diff --git a/config/install.yaml b/config/install.yaml
@@ -6170,8 +6170,14 @@ spec:
                       type: string
                     scale:
                       properties:
+                        lookbackSeconds:
+                          default: 180
+                          description: Lookback seconds to calculate the average pending
+                            messages and processing rate
+                          format: int32
+                          type: integer
                         max:
-                          default: 1
+                          default: 100
                           description: Maximum replicas
                           format: int32
                           type: integer
@@ -10057,8 +10063,14 @@ spec:
                 type: integer
               scale:
                 properties:
+                  lookbackSeconds:
+                    default: 180
+                    description: Lookback seconds to calculate the average pending
+                      messages and processing rate
+                    format: int32
+                    type: integer
                   max:
-                    default: 1
+                    default: 100
                     description: Maximum replicas
                     format: int32
                     type: integer

diff --git a/docs/APIs.md b/docs/APIs.md
@@ -2983,6 +2983,18 @@ Maximum replicas
 </p>
 </td>
 </tr>
+<tr>
+<td>
+<code>lookbackSeconds</code></br> <em> int32 </em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>
+Lookback seconds to calculate the average pending messages and
+processing rate
+</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="numaflow.numaproj.io/v1alpha1.Sink">
@@ -3273,6 +3285,10 @@ Description
 Vertex
 </h3>
 <p>
+(<em>Appears on:</em>
+<a href="#numaflow.numaproj.io/v1alpha1.VertexInstance">VertexInstance</a>)
+</p>
+<p>
 </p>
 <table>
 <thead>
@@ -3374,6 +3390,52 @@ Refer to the Kubernetes API documentation for the fields of the
 </tr>
 </tbody>
 </table>
+<h3 id="numaflow.numaproj.io/v1alpha1.VertexInstance">
+VertexInstance
+</h3>
+<p>
+<p>
+VertexInstance is a wrapper of a vertex instance, which contains the
+vertex spec and the instance information such as hostname and replica
+index.
+</p>
+</p>
+<table>
+<thead>
+<tr>
+<th>
+Field
+</th>
+<th>
+Description
+</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>vertex</code></br> <em>
+<a href="#numaflow.numaproj.io/v1alpha1.Vertex"> Vertex </a> </em>
+</td>
+<td>
+</td>
+</tr>
+<tr>
+<td>
+<code>hostname</code></br> <em> string </em>
+</td>
+<td>
+</td>
+</tr>
+<tr>
+<td>
+<code>replica</code></br> <em> int32 </em>
+</td>
+<td>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="numaflow.numaproj.io/v1alpha1.VertexLimits">
 VertexLimits
 </h3>