Repository: numaproj/numaflow

Issue:
Label for isb_*_error_total metrics should be "buffer"
It's an issue related to the following four metrics:

- isb_jetstream_read_error_total
- isb_jetstream_write_error_total
- isb_redis_read_error_total
- isb_redis_write_error_total

The Operator Manual [Metrics](https://numaflow.numaproj.io/operations/metrics/metrics/) says that "Labels" of the four are `partition_name=<partition-name>`. But in the Golang codes, their "[labelNames](https://pkg.go.dev/github.com/prometheus/client_golang/prometheus/promauto#NewCounterVec)" are all `"buffer"`:

https://github.com/numaproj/numaflow/blob/19b57d2b2996e5d4b8254832c7eef764469f691a/pkg/isb/stores/jetstream/metrics.go#L24-L29

https://github.com/numaproj/numaflow/blob/19b57d2b2996e5d4b8254832c7eef764469f691a/pkg/isb/stores/jetstream/metrics.go#L45-L50

https://github.com/numaproj/numaflow/blob/19b57d2b2996e5d4b8254832c7eef764469f691a/pkg/isb/stores/redis/metrics.go#L31-L36

https://github.com/numaproj/numaflow/blob/19b57d2b2996e5d4b8254832c7eef764469f691a/pkg/isb/stores/redis/metrics.go#L59-L64

I think It's a documentation bug. `buffer=<buffer-name>` looks correct.


Discussion:

Solution patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 3ef0c941..e02969a2 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -217,6 +217,7 @@ jobs:
             idle-source-e2e,
             monovertex-e2e,
             builtin-source-e2e,
+            serving-e2e,
           ]
         include:
           - driver: redis
diff --git a/docs/operations/metrics/metrics.md b/docs/operations/metrics/metrics.md
index 35194f3e..4fd1bc22 100644
--- a/docs/operations/metrics/metrics.md
+++ b/docs/operations/metrics/metrics.md
@@ -64,10 +64,10 @@ These metrics can be used to determine if there are any errors in the pipeline.
 | `forwarder_fbsink_write_error_total`       | Counter     | `pipeline=<pipeline-name>` <br> `vertex=<vertex-name>` `vertex_type=<vertex-type>` <br> <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Indicates any errors while writing to a fallback sink                                           |
 | `forwarder_ack_error_total`                | Counter     | `pipeline=<pipeline-name>` <br> `vertex=<vertex-name>` <br> `vertex_type=<vertex-type>` <br> `replica=<replica-index>` <br> `partition_name=<partition-name>` | Indicates any errors while acknowledging messages by the forwarder                              |
 | `kafka_sink_write_timeout_total`           | Counter     | `pipeline=<pipeline-name>` <br> `vertex=<vertex-name>`                                                                                                        | Provides the write timeouts while writing to the Kafka sink                                     |
-| `isb_jetstream_read_error_total`           | Counter     | `partition_name=<partition-name>`                                                                                                                             | Indicates any read errors with NATS Jetstream ISB                                               |
-| `isb_jetstream_write_error_total`          | Counter     | `partition_name=<partition-name>`                                                                                                                             | Indicates any write errors with NATS Jetstream ISB                                              |
-| `isb_redis_read_error_total`               | Counter     | `partition_name=<partition-name>`                                                                                                                             | Indicates any read errors with Redis ISB                                                        |
-| `isb_redis_write_error_total`              | Counter     | `partition_name=<partition-name>`                                                                                                                             | Indicates any write errors with Redis ISB                                                       |
+| `isb_jetstream_read_error_total`           | Counter     | `buffer=<buffer-name>`                                                                                                                                        | Indicates any read errors with NATS Jetstream ISB                                               |
+| `isb_jetstream_write_error_total`          | Counter     | `buffer=<buffer-name>`                                                                                                                                        | Indicates any write errors with NATS Jetstream ISB                                              |
+| `isb_redis_read_error_total`               | Counter     | `buffer=<buffer-name>`                                                                                                                                        | Indicates any read errors with Redis ISB                                                        |
+| `isb_redis_write_error_total`              | Counter     | `buffer=<buffer-name>`                                                                                                                                        | Indicates any write errors with Redis ISB                                                       |
 
 ### Saturation
 



Test patch:
diff --git a/test/e2e-api/main.go b/test/e2e-api/main.go
index e15cf09a..c7092ca8 100644
--- a/test/e2e-api/main.go
+++ b/test/e2e-api/main.go
@@ -39,6 +39,10 @@ func main() {
 	httpController := NewHttpController()
 	http.HandleFunc("/http/send-message", httpController.SendMessage)
 
+	servingController := NewServingController()
+	http.HandleFunc("/serving/send-message", servingController.SendMessage)
+	http.HandleFunc("/serving/fetch-results", servingController.FetchResults)
+
 	// initialize NATS handler
 	natsController := NewNatsController("nats", "testingtoken")
 	http.HandleFunc("/nats/pump-subject", natsController.PumpSubject)
diff --git a/test/e2e-api/serving.go b/test/e2e-api/serving.go
new file mode 100644
index 00000000..363ef924
--- /dev/null
+++ b/test/e2e-api/serving.go
@@ -0,0 +1,133 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"bytes"
+	"crypto/tls"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+)
+
+type ServingController struct {
+	client *http.Client
+}
+
+func NewServingController() *ServingController {
+	return &ServingController{
+		client: &http.Client{
+			Transport: &http.Transport{
+				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+			},
+		},
+	}
+}
+
+func (h *ServingController) SendMessage(w http.ResponseWriter, r *http.Request) {
+	host := r.URL.Query().Get("host")
+	sync := r.URL.Query().Get("sync")
+	reqId := r.URL.Query().Get("reqId")
+	reqBytes, err := io.ReadAll(r.Body)
+	if err != nil {
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	uri := fmt.Sprintf("https://%s:8443/v1/process/sync", host)
+	if sync == "false" {
+		uri = fmt.Sprintf("https://%s:8443/v1/process/async", host)
+	}
+
+	postReq, err := http.NewRequest("POST", uri, bytes.NewBuffer(reqBytes))
+	if err != nil {
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if reqId != "" {
+		postReq.Header.Set("X-Numaflow-Id", reqId)
+	}
+
+	resp, err := h.client.Do(postReq)
+	if err != nil {
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		log.Printf("[ERROR] Expected %d, Got %d", http.StatusOK, resp.StatusCode)
+		http.Error(w, fmt.Sprintf("Bad status: %s", resp.Status), http.StatusInternalServerError)
+		return
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		log.Printf("[ERROR] Reading response body from Serving source: %v", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	_, _ = w.Write(body)
+}
+
+func (h *ServingController) FetchResults(w http.ResponseWriter, r *http.Request) {
+	host := r.URL.Query().Get("host")
+	reqId := r.URL.Query().Get("reqId")
+
+	if reqId == "" {
+		http.Error(w, "request id can not be empty for fetch API", http.StatusInternalServerError)
+		return
+	}
+
+	uri := fmt.Sprintf("https://%s:8443/v1/process/fetch?id=%s", host, reqId)
+
+	req, err := http.NewRequest("GET", uri, nil)
+	if err != nil {
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	resp, err := h.client.Do(req)
+	if err != nil {
+		log.Println(err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		log.Printf("[ERROR] Expected %d, Got %d", http.StatusOK, resp.StatusCode)
+		http.Error(w, fmt.Sprintf("Bad status: %s", resp.Status), http.StatusInternalServerError)
+		return
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		log.Printf("[ERROR] Reading response body from Serving source: %v", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	_, _ = w.Write(body)
+}
+
+// Close closes the http client
+func (h *ServingController) Close() {
+	h.client.CloseIdleConnections()
+}
diff --git a/test/fixtures/e2eapi.go b/test/fixtures/e2eapi.go
index c9e92f2b..0351e38e 100644
--- a/test/fixtures/e2eapi.go
+++ b/test/fixtures/e2eapi.go
@@ -72,7 +72,8 @@ func InvokeE2EAPIPOST(format string, body string, args ...interface{}) string {
 		if err == nil && resp.StatusCode < 300 {
 			return true, nil
 		}
-		fmt.Printf("Got error %v, response %v, retrying.\n", err, *resp)
+		body, _ := io.ReadAll(resp.Body)
+		log.Printf("Got error %v, response_text: %s,response: %+v, retrying.\n", err, body, *resp)
 		return false, nil
 	})
 
@@ -84,16 +85,17 @@ func InvokeE2EAPIPOST(format string, body string, args ...interface{}) string {
 		_ = Body.Close()
 	}(resp.Body)
 
+	var respBody string
 	for s := bufio.NewScanner(resp.Body); s.Scan(); {
 		x := s.Text()
 		if strings.Contains(x, "ERROR") { // hacky way to return an error from an octet-stream
 			panic(errors.New(x))
 		}
 		log.Printf("> %s\n", x)
-		body += x
+		respBody += x
 	}
 	if resp.StatusCode >= 300 {
 		panic(errors.New(resp.Status))
 	}
-	return body
+	return respBody
 }
diff --git a/test/fixtures/serving.go b/test/fixtures/serving.go
new file mode 100644
index 00000000..1209a81c
--- /dev/null
+++ b/test/fixtures/serving.go
@@ -0,0 +1,25 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package fixtures
+
+func SendServingMessage(host, reqId, payload string, sync bool) string {
+	return InvokeE2EAPIPOST("/serving/send-message?host=%s&reqId=%s&sync=%t", payload, host, reqId, sync)
+}
+
+func FetchServingResult(host, reqId string) string {
+	return InvokeE2EAPIPOST("/serving/fetch-results?host=%s&reqId=%s", "", host, reqId)
+}
diff --git a/test/serving-e2e/serving_test.go b/test/serving-e2e/serving_test.go
new file mode 100644
index 00000000..590f7491
--- /dev/null
+++ b/test/serving-e2e/serving_test.go
@@ -0,0 +1,112 @@
+//go:build test
+
+/*
+Copyright 2022 The Numaproj Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package serving_e2e
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/numaproj/numaflow/test/fixtures"
+)
+
+//go:generate kubectl -n numaflow-system set env deploy/numaflow-controller NUMAFLOW_EXECUTE_RUST_BINARY=true
+//go:generate kubectl delete -f testdata/serving-pipeline-cat.yaml -n numaflow-system --ignore-not-found=true
+//go:generate kubectl -n numaflow-system wait --for=condition=ready pod -l app.kubernetes.io/name=controller-manager --timeout 30s
+
+const Namespace = "numaflow-system"
+
+type ServingSuite struct {
+	fixtures.E2ESuite
+}
+
+// {"message":"Successfully published message","id":"0195859e-b05f-78d3-b0b6-6946a77f842e","code":200,"timestamp":"2025-03-11T14:32:05.455629106Z"}
+type asyncAPIResponse struct {
+	Message   string    `json:"message"`
+	Id        string    `json:"id"`
+	Code      int       `json:"code"`
+	Timestamp time.Time `json:"timestamp"`
+}
+
+func (resp *asyncAPIResponse) validate(expReqId string) error {
+	if resp.Message != "Successfully published message" {
+		return fmt.Errorf("message field = %q, expected='Successfully published message'", resp.Message)
+	}
+	if resp.Id != expReqId {
+		return fmt.Errorf("value of id field should be %q, current value: %q", expReqId, resp.Id)
+	}
+	var defaultTime time.Time
+	if resp.Timestamp == defaultTime {
+		return fmt.Errorf("invalid value for timestamp field: %q", resp.Timestamp)
+	}
+	if resp.Code != 200 {
+		return fmt.Errorf("expected value of 'code' field is 200, got %d", resp.Code)
+	}
+	return nil
+}
+
+func (ss *ServingSuite) TestServingSource() {
+	w := ss.Given().Pipeline("@testdata/serving-pipeline-cat.yaml").
+		When().
+		CreatePipelineAndWait()
+	defer w.DeletePipelineAndWait()
+
+	w.Expect().VertexPodsRunning()
+
+	pipelineName := "serving-source"
+	serviceName := "serving-source-serving-in"
+	// Check Service
+	cmd := fmt.Sprintf("kubectl -n %s get svc -lnumaflow.numaproj.io/pipeline-name=%s,numaflow.numaproj.io/vertex-name=%s | grep -v CLUSTER-IP | grep -v headless", Namespace, pipelineName, "serving-in")
+	w.Exec("sh", []string{"-c", cmd}, fixtures.OutputRegexp(serviceName))
+
+	// Send a request using sync API
+	syncResp := fixtures.SendServingMessage(serviceName, "", "test data", true)
+	assert.Equal(ss.T(), "test data", syncResp)
+
+	// Send a request using async API
+	const reqId = "req-12345"
+	asyncRespText := fixtures.SendServingMessage(serviceName, reqId, "test data", false)
+	var asyncResp asyncAPIResponse
+	err := json.Unmarshal([]byte(asyncRespText), &asyncResp)
+	require.NoError(ss.T(), err)
+	require.NoError(ss.T(), asyncResp.validate(reqId))
+
+	// Use fetch API to retrieve the results of the above async request
+	attempt := 0
+	var asyncResult string
+	for {
+		if attempt > 5 {
+			ss.T().Fatalf("Processing for async request is still in-progress")
+			break
+		}
+		asyncResult = fixtures.FetchServingResult(serviceName, reqId)
+		if asyncResult != `{"status":"in-progress"}` {
+			break
+		}
+		attempt++
+		time.Sleep(2 * time.Second)
+	}
+	assert.Equal(ss.T(), "test data", asyncResult)
+}
+
+func TestServingSuite(t *testing.T) {
+	suite.Run(t, new(ServingSuite))
+}
diff --git a/test/serving-e2e/testdata/serving-pipeline-cat.yaml b/test/serving-e2e/testdata/serving-pipeline-cat.yaml
new file mode 100644
index 00000000..344ab5ed
--- /dev/null
+++ b/test/serving-e2e/testdata/serving-pipeline-cat.yaml
@@ -0,0 +1,39 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: serving-source
+spec:
+  vertices:
+    - name: serving-in
+      servingStoreName: default
+      scale:
+        min: 1
+        max: 1
+      source:
+        serving:
+          service: true
+          msgIDHeaderKey: "X-Numaflow-Id"
+
+    - name: cat
+      scale:
+        min: 1
+        max: 1
+      udf:
+        container:
+          image: quay.io/numaio/numaflow-go/map-forward-message:stable
+
+    - name: serve-sink
+      servingStoreName: default
+      scale:
+        min: 1
+        max: 1
+      sink:
+        udsink:
+          container:
+            image: quay.io/numaio/numaflow-go/sink-serve:stable
+
+  edges:
+    - from: serving-in
+      to: cat
+    - from: cat
+      to: serve-sink
\ No newline at end of file