Repository: numaproj/numaflow
Pull request #81 (2022-06-29 05:43:37)

chore: added yarn build to docker file
In the below PR we have added the following:
1. Dockerfile: Added yarn build to the docker image and copied the UI in the main numaflow binary. (I did not research if there was a way to compress this yet)
2. The image size prior to adding UI director `ui/build` is:
<img width="808" alt="Screen Shot 2022-06-28 at 8 48 37 PM" src="https://user-images.githubusercontent.com/11369796/176354049-61699cff-5765-43e6-ac07-77ea6f6ea2db.png">
3. The image size after adding UI director `ui/build` is:
<img width="769" alt="Screen Shot 2022-06-28 at 9 14 51 PM" src="https://user-images.githubusercontent.com/11369796/176354105-3fdd3684-42ce-444f-b853-c1bbaf29cb49.png">
4. The make start (which runs make image in turn runs docker-build) command with cached package.json changes and yarn build takes the following time:
<img width="1558" alt="Screen Shot 2022-06-28 at 10 41 58 PM" src="https://user-images.githubusercontent.com/11369796/176360374-3909c7a0-ffdf-4e9b-85ed-4b5ae9cc6c75.png">
5.  I think this is reasonable and we could take a run at the release github action to see how it behaves.
6. running `kubectl apply -f ./config/install.yaml` now installs the numaflow-server too. (I have not updated the README.md yet, we could take a stab at it if this approach is ok)

whynowy: I can see the time increase is about 2 mins for each image build.

Patch:
diff --git a/Dockerfile b/Dockerfile
index bad384b6..8b7ac5bf 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,16 @@
 ARG ARCH=$TARGETARCH
+####################################################################################################
+FROM docker.io/node:18 as numaflow-ui
+
+WORKDIR /app
+
+COPY ui/package.json ui/yarn.lock ui/
+
+RUN JOBS=max yarn --cwd ui install --network-timeout 1000000
+
+COPY ui ui
+RUN NODE_OPTIONS="--max-old-space-size=2048" JOBS=max yarn --cwd ui build
+
 ####################################################################################################
 # base
 ####################################################################################################
@@ -19,6 +31,7 @@ ARG ARCH
 COPY --from=base /usr/share/zoneinfo /usr/share/zoneinfo
 COPY --from=base /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
 COPY --from=base /bin/numaflow /bin/numaflow
+COPY --from=numaflow-ui /app/ui/build /ui/build
 ENTRYPOINT [ "/bin/numaflow" ]
 
 ####################################################################################################
diff --git a/config/base/kustomization.yaml b/config/base/kustomization.yaml
index 47eb09e4..9cbb9e89 100644
--- a/config/base/kustomization.yaml
+++ b/config/base/kustomization.yaml
@@ -4,7 +4,9 @@ kind: Kustomization
 resources:
   - crds
   - numaflow-sa.yaml
+  - numaflow-server-sa.yaml
   - controller-manager
+  - numaflow-server
 
 images:
   - name: quay.io/numaproj/numaflow
@@ -24,3 +26,17 @@ patchesStrategicMerge:
               env:
                 - name: NUMAFLOW_IMAGE
                   value: quay.io/numaproj/numaflow:latest
+  - |-
+    apiVersion: apps/v1
+    kind: Deployment
+    metadata:
+      name: numaflow-server
+    spec:
+      template:
+        spec:
+          containers:
+            - name: main
+              image: quay.io/numaproj/numaflow:latest
+              args: 
+              - "server"
+
diff --git a/config/base/numaflow-server-sa.yaml b/config/base/numaflow-server-sa.yaml
new file mode 100644
index 00000000..6af0a8fd
--- /dev/null
+++ b/config/base/numaflow-server-sa.yaml
@@ -0,0 +1,4 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: numaflow-server-sa
diff --git a/config/base/numaflow-server/kustomization.yaml b/config/base/numaflow-server/kustomization.yaml
new file mode 100644
index 00000000..66a906af
--- /dev/null
+++ b/config/base/numaflow-server/kustomization.yaml
@@ -0,0 +1,6 @@
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+
+resources:
+  - numaflow-server-deployment.yaml
+  - numaflow-server-service.yaml
\ No newline at end of file
diff --git a/config/base/numaflow-server/numaflow-server-deployment.yaml b/config/base/numaflow-server/numaflow-server-deployment.yaml
new file mode 100644
index 00000000..0c15fb6a
--- /dev/null
+++ b/config/base/numaflow-server/numaflow-server-deployment.yaml
@@ -0,0 +1,45 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: numaflow-server
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/part-of: numaflow
+      app.kubernetes.io/component: numaflow-ux
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/part-of: numaflow
+        app.kubernetes.io/component: numaflow-ux
+    spec:
+      serviceAccountName: numaflow-server-sa
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 9737
+      containers:
+        - name: main
+          image: quay.io/numaproj/numaflow:latest
+          args:
+          - "server"
+          imagePullPolicy: Always
+          env:
+            - name: NAMESPACE
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.namespace
+          resources:
+            limits:
+              cpu: 500m
+              memory: 1024Mi
+            requests:
+              cpu: 100m
+              memory: 200Mi
+          livenessProbe:
+            httpGet:
+              path: /healthz
+              port: 8443
+              scheme: HTTPS
+            initialDelaySeconds: 3
+            periodSeconds: 3
diff --git a/config/base/numaflow-server/numaflow-server-service.yaml b/config/base/numaflow-server/numaflow-server-service.yaml
new file mode 100644
index 00000000..25f5143a
--- /dev/null
+++ b/config/base/numaflow-server/numaflow-server-service.yaml
@@ -0,0 +1,12 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: numaflow-server-svc
+spec:
+  type: ClusterIP
+  ports:
+  - targetPort: 8443
+    port: 8443
+  selector:
+    app.kubernetes.io/part-of: numaflow
+    app.kubernetes.io/component: numaflow-ux
\ No newline at end of file
diff --git a/config/cluster-install/rbac/kustomization.yaml b/config/cluster-install/rbac/kustomization.yaml
index 0b2a48f5..8c7cba84 100644
--- a/config/cluster-install/rbac/kustomization.yaml
+++ b/config/cluster-install/rbac/kustomization.yaml
@@ -7,3 +7,5 @@ resources:
   - numaflow-aggregate-to-view.yaml
   - numaflow-cluster-role.yaml
   - numaflow-binding.yaml
+  - numaflow-server-binding.yaml
+  - numaflow-server-cluster-role.yaml
\ No newline at end of file
diff --git a/config/cluster-install/rbac/numaflow-server-binding.yaml b/config/cluster-install/rbac/numaflow-server-binding.yaml
new file mode 100644
index 00000000..5743f7fd
--- /dev/null
+++ b/config/cluster-install/rbac/numaflow-server-binding.yaml
@@ -0,0 +1,12 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: numaflow-server-binding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: numaflow-server-role
+subjects:
+  - kind: ServiceAccount
+    name: numaflow-server-sa
+    namespace: numaflow-system
\ No newline at end of file
diff --git a/config/cluster-install/rbac/numaflow-server-cluster-role.yaml b/config/cluster-install/rbac/numaflow-server-cluster-role.yaml
new file mode 100644
index 00000000..500489f3
--- /dev/null
+++ b/config/cluster-install/rbac/numaflow-server-cluster-role.yaml
@@ -0,0 +1,52 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: numaflow-server-role
+rules:
+  - apiGroups:
+      - numaflow.numaproj.io
+    verbs:
+      - get
+      - list
+      - watch
+    resources:
+      - interstepbufferservices
+      - interstepbufferservices/finalizers
+      - interstepbufferservices/status
+      - pipelines
+      - pipelines/finalizers
+      - pipelines/status
+      - vertices
+      - vertices/finalizers
+      - vertices/status
+      - vertices/scale
+  - apiGroups:
+      - ""
+    resources:
+      - pods
+      - pods/exec
+      - pods/log
+      - configmaps
+      - secrets
+      - services
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+      - statefulsets
+    verbs:
+      - get
+      - list
+      - watch
+  - apiGroups:
+      - metrics.k8s.io
+    resources:
+      - pods
+    verbs:
+      - get
+      - list
+      - watch
diff --git a/config/install.yaml b/config/install.yaml
index 9ff347ee..f10a9863 100644
--- a/config/install.yaml
+++ b/config/install.yaml
@@ -12491,6 +12491,12 @@ metadata:
   name: numaflow-sa
   namespace: numaflow-system
 ---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: numaflow-server-sa
+  namespace: numaflow-system
+---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
@@ -12647,6 +12653,59 @@ rules:
   - delete
 ---
 apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: numaflow-server-role
+rules:
+- apiGroups:
+  - numaflow.numaproj.io
+  resources:
+  - interstepbufferservices
+  - interstepbufferservices/finalizers
+  - interstepbufferservices/status
+  - pipelines
+  - pipelines/finalizers
+  - pipelines/status
+  - vertices
+  - vertices/finalizers
+  - vertices/status
+  - vertices/scale
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  - pods/exec
+  - pods/log
+  - configmaps
+  - secrets
+  - services
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - apps
+  resources:
+  - deployments
+  - statefulsets
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - metrics.k8s.io
+  resources:
+  - pods
+  verbs:
+  - get
+  - list
+  - watch
+---
+apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
 metadata:
   name: numaflow-binding
@@ -12659,6 +12718,19 @@ subjects:
   name: numaflow-sa
   namespace: numaflow-system
 ---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: numaflow-server-binding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: numaflow-server-role
+subjects:
+- kind: ServiceAccount
+  name: numaflow-server-sa
+  namespace: numaflow-system
+---
 apiVersion: v1
 data:
   controller-config.yaml: |
@@ -12767,6 +12839,20 @@ metadata:
   name: numaflow-controller-config
   namespace: numaflow-system
 ---
+apiVersion: v1
+kind: Service
+metadata:
+  name: numaflow-server-svc
+  namespace: numaflow-system
+spec:
+  ports:
+  - port: 8443
+    targetPort: 8443
+  selector:
+    app.kubernetes.io/component: numaflow-ux
+    app.kubernetes.io/part-of: numaflow
+  type: ClusterIP
+---
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -12827,3 +12913,50 @@ spec:
       - configMap:
           name: numaflow-controller-config
         name: controller-config-volume
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: numaflow-server
+  namespace: numaflow-system
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app.kubernetes.io/component: numaflow-ux
+      app.kubernetes.io/part-of: numaflow
+  template:
+    metadata:
+      labels:
+        app.kubernetes.io/component: numaflow-ux
+        app.kubernetes.io/part-of: numaflow
+    spec:
+      containers:
+      - args:
+        - server
+        env:
+        - name: NAMESPACE
+          valueFrom:
+            fieldRef:
+              fieldPath: metadata.namespace
+        image: quay.io/numaproj/numaflow:latest
+        imagePullPolicy: Always
+        livenessProbe:
+          httpGet:
+            path: /healthz
+            port: 8443
+            scheme: HTTPS
+          initialDelaySeconds: 3
+          periodSeconds: 3
+        name: main
+        resources:
+          limits:
+            cpu: 500m
+            memory: 1024Mi
+          requests:
+            cpu: 100m
+            memory: 200Mi
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 9737
+      serviceAccountName: numaflow-server-sa
diff --git a/pkg/sources/kafka/reader.go b/pkg/sources/kafka/reader.go
index 0b6aeee0..59c0edae 100644
--- a/pkg/sources/kafka/reader.go
+++ b/pkg/sources/kafka/reader.go
@@ -15,6 +15,7 @@ import (
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"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/shared/util"
 	"github.com/numaproj/numaflow/pkg/udf/applier"
 )
@@ -25,7 +26,7 @@ type KafkaSource struct {
 	// name of the pipeline
 	pipelineName string
 	// group name for the source vertex
-	groupname string
+	groupName string
 	// topic to consume messages from
 	topic string
 	// kafka brokers
@@ -48,6 +49,10 @@ type KafkaSource struct {
 	handlerbuffer int
 	// read timeout for the from buffer
 	readTimeout time.Duration
+	// client used to calculate pending messages
+	adminClient sarama.ClusterAdmin
+	// sarama client
+	saramaClient sarama.Client
 }
 
 type Option func(*KafkaSource) error
@@ -79,7 +84,7 @@ func WithReadTimeOut(t time.Duration) Option {
 // WithGroupName is used to set the group name
 func WithGroupName(gn string) Option {
 	return func(o *KafkaSource) error {
-		o.groupname = gn
+		o.groupName = gn
 		return nil
 	}
 }
@@ -160,14 +165,43 @@ func (r *KafkaSource) Close() error {
 	r.logger.Info("Closing kafka reader...")
 	// finally, shut down the client
 	r.cancelfn()
+	if r.adminClient != nil {
+		if err := r.adminClient.Close(); err != nil {
+			r.logger.Errorw("Error in closing kafka admin client", zap.Error(err))
+		}
+	}
+	if r.saramaClient != nil {
+		if err := r.saramaClient.Close(); err != nil {
+			r.logger.Errorw("Error in closing kafka sarama client", zap.Error(err))
+		}
+	}
 	<-r.stopch
 	r.logger.Info("Kafka reader closed")
 	return nil
 }
 
 func (r *KafkaSource) Pending(ctx context.Context) (int64, error) {
-	// TODO: not implemented
-	return isb.PendingNotAvailable, nil
+	if r.adminClient == nil || r.saramaClient == nil {
+		return isb.PendingNotAvailable, nil
+	}
+	partitions, err := r.saramaClient.Partitions(r.topic)
+	if err != nil {
+		return isb.PendingNotAvailable, fmt.Errorf("failed to get partitions, %w", err)
+	}
+	totalPending := int64(0)
+	rep, err := r.adminClient.ListConsumerGroupOffsets(r.groupName, map[string][]int32{r.topic: partitions})
+	if err != nil {
+		return isb.PendingNotAvailable, fmt.Errorf("failed to list consumer group offsets, %w", err)
+	}
+	for _, partition := range partitions {
+		partitionOffset, err := r.saramaClient.GetOffset(r.topic, partition, sarama.OffsetNewest)
+		if err != nil {
+			return isb.PendingNotAvailable, fmt.Errorf("failed to get offset of topic %q, partition %v, %w", r.topic, partition, err)
+		}
+		block := rep.GetBlock(r.topic, partition)
+		totalPending += partitionOffset - block.Offset
+	}
+	return totalPending, nil
 }
 
 // NewKafkaSource returns a KafkaSource reader based on Kafka Consumer Group .
@@ -178,8 +212,9 @@ func NewKafkaSource(vertex *dfv1.Vertex, writers []isb.BufferWriter, opts ...Opt
 		pipelineName:  vertex.Spec.PipelineName,
 		topic:         source.Topic,
 		brokers:       source.Brokers,
-		readTimeout:   1 * time.Second, // default timeout
-		handlerbuffer: 100,             // default buffer size for kafka reads
+		readTimeout:   1 * time.Second,     // default timeout
+		handlerbuffer: 100,                 // default buffer size for kafka reads
+		logger:        logging.NewLogger(), // default logger
 	}
 
 	for _, o := range opts {
@@ -202,8 +237,20 @@ func NewKafkaSource(vertex *dfv1.Vertex, writers []isb.BufferWriter, opts ...Opt
 			config.Net.TLS.Config = c
 		}
 	}
-
 	kafkasource.config = config
+	// Best effort to initialize the clients for pending messages calculation
+	adminClient, err := sarama.NewClusterAdmin(kafkasource.brokers, config)
+	if err != nil {
+		kafkasource.logger.Warnw("Problem initializing sarama admin client", zap.Error(err))
+	} else {
+		kafkasource.adminClient = adminClient
+	}
+	client, err := sarama.NewClient(kafkasource.brokers, config)
+	if err != nil {
+		kafkasource.logger.Warnw("Problem initializing sarama client", zap.Error(err))
+	} else {
+		kafkasource.saramaClient = client
+	}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	kafkasource.cancelfn = cancel
@@ -231,24 +278,21 @@ func NewKafkaSource(vertex *dfv1.Vertex, writers []isb.BufferWriter, opts ...Opt
 		return nil, err
 	}
 	kafkasource.forwarder = forwarder
-
 	return kafkasource, nil
 }
 
 func configFromOpts(yamlconfig string) (*sarama.Config, error) {
-
 	config, err := util.GetSaramaConfigFromYAMLString(yamlconfig)
 	if err != nil {
 		return nil, err
 	}
 	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
-
 	return config, nil
 }
 
 func (r *KafkaSource) startConsumer() {
-	client, err := sarama.NewConsumerGroup(r.brokers, r.groupname, r.config)
-	r.logger.Infow("creating NewConsumerGroup", zap.String("topic", r.topic), zap.String("consumerGroupName", r.groupname), zap.Strings("brokers", r.brokers))
+	client, err := sarama.NewConsumerGroup(r.brokers, r.groupName, r.config)
+	r.logger.Infow("creating NewConsumerGroup", zap.String("topic", r.topic), zap.String("consumerGroupName", r.groupName), zap.Strings("brokers", r.brokers))
 	if err != nil {
 		r.logger.Panicw("Problem initializing sarama client", zap.Error(err))
 	}
@@ -275,7 +319,6 @@ func (r *KafkaSource) startConsumer() {
 }
 
 func toReadMessage(m *sarama.ConsumerMessage) *isb.ReadMessage {
-
 	offset := toOffset(m.Topic, m.Partition, m.Offset)
 	msg := isb.Message{
 		Header: isb.Header{
diff --git a/ui/src/components/pipeline/graph/Graph.tsx b/ui/src/components/pipeline/graph/Graph.tsx
index 661c59cc..0d748010 100644
--- a/ui/src/components/pipeline/graph/Graph.tsx
+++ b/ui/src/components/pipeline/graph/Graph.tsx
@@ -131,7 +131,7 @@ export default function Graph(props: GraphProps) {
     setEdgeOpen(false);
   };
 
-  const handlePaneClick = (event: MouseEvent) => {
+  const handlePaneClick = () => {
     setShowSpec(true);
     setEdgeOpen(false);
     setNodeOpen(false);
@@ -165,9 +165,8 @@ export default function Graph(props: GraphProps) {
           onNodeClick={handleNodeClick}
           onPaneClick={handlePaneClick}
           fitView
-          zoomOnScroll={false}
-          panOnDrag={false}
-          elevateEdgesOnSelect={true}
+          zoomOnScroll={true}
+          panOnDrag={true}
         />
       </div>
 
@@ -189,4 +188,4 @@ export default function Graph(props: GraphProps) {
       </Card>
     </div>
   );
-}
+}
\ No newline at end of file


Test patch:
diff --git a/pkg/sources/kafka/reader_test.go b/pkg/sources/kafka/reader_test.go
index 80729b16..04618e4f 100644
--- a/pkg/sources/kafka/reader_test.go
+++ b/pkg/sources/kafka/reader_test.go
@@ -31,7 +31,7 @@ func TestNewKafkasource(t *testing.T) {
 	assert.Nil(t, err)
 	assert.NotNil(t, ks)
 
-	assert.Equal(t, "default", ks.groupname)
+	assert.Equal(t, "default", ks.groupName)
 
 	// config is all set and initialized correctly
 	assert.NotNil(t, ks.config)
@@ -56,7 +56,7 @@ func TestGroupNameOverride(t *testing.T) {
 	}}
 	ks, _ := NewKafkaSource(vertex, dest, WithLogger(logging.NewLogger()), WithBufferSize(100), WithReadTimeOut(100*time.Millisecond), WithGroupName("default"))
 
-	assert.Equal(t, "default", ks.groupname)
+	assert.Equal(t, "default", ks.groupName)
 
 }
 
diff --git a/test/manifests/kustomization.yaml b/test/manifests/kustomization.yaml
index e4fdd29c..098e4cce 100644
--- a/test/manifests/kustomization.yaml
+++ b/test/manifests/kustomization.yaml
@@ -21,6 +21,13 @@ patches:
     target:
       kind: Deployment
       name: controller-manager
+  - patch: |-
+      - op: add
+        path: /spec/template/spec/containers/0/imagePullPolicy
+        value: IfNotPresent
+    target:
+      kind: Deployment
+      name: numaflow-server
 namespace: numaflow-system
 
 commonLabels: