diff --git a/api/json-schema/schema.json b/api/json-schema/schema.json index 778f51e585..0b3c4eeecb 100644 --- a/api/json-schema/schema.json +++ b/api/json-schema/schema.json @@ -17858,7 +17858,7 @@ "type": "object" }, "io.numaproj.numaflow.v1alpha1.Container": { - "description": "Container is used to define the container properties for user defined functions, sinks, etc.", + "description": "Container is used to define the container properties for user-defined functions, sinks, etc.", "properties": { "args": { "items": { diff --git a/api/openapi-spec/swagger.json b/api/openapi-spec/swagger.json index ba645cfd04..61d18a8711 100644 --- a/api/openapi-spec/swagger.json +++ b/api/openapi-spec/swagger.json @@ -17862,7 +17862,7 @@ } }, "io.numaproj.numaflow.v1alpha1.Container": { - "description": "Container is used to define the container properties for user defined functions, sinks, etc.", + "description": "Container is used to define the container properties for user-defined functions, sinks, etc.", "type": "object", "properties": { "args": { diff --git a/docs/APIs.md b/docs/APIs.md index b44d7eeeb2..766a56cb95 100644 --- a/docs/APIs.md +++ b/docs/APIs.md @@ -791,7 +791,7 @@ Container

-Container is used to define the container properties for user defined +Container is used to define the container properties for user-defined functions, sinks, etc.

diff --git a/docs/core-concepts/vertex.md b/docs/core-concepts/vertex.md index fe5ae85745..5feed6b7bc 100644 --- a/docs/core-concepts/vertex.md +++ b/docs/core-concepts/vertex.md @@ -6,7 +6,7 @@ There are 3 types of `Vertex` in Numaflow today: 1. `Source` - To ingest data from sources. 1. `Sink` - To forward processed data to sinks. -1. `UDF` - User Defined Function, which is used to define data processing logic. +1. `UDF` - User-defined Function, which is used to define data processing logic. We have defined a [Kubernetes Custom Resource](https://kubernetes.io/docs/concepts/extend-kubernetes/api-extension/custom-resources/) for `Vertex`. A `Pipeline` containing multiple vertices will automatically generate multiple `Vertex` objects by the controller. As a user, you should NOT create a `Vertex` object directly. diff --git a/docs/core-concepts/watermarks.md b/docs/core-concepts/watermarks.md index d80cde7e9d..6f9d61d739 100644 --- a/docs/core-concepts/watermarks.md +++ b/docs/core-concepts/watermarks.md @@ -70,7 +70,7 @@ spec: ## Watermark API -When processing data in [User Defined Functions](../user-guide/user-defined-functions/user-defined-functions.md), you can get the current watermark through +When processing data in [user-defined functions](../user-guide/user-defined-functions/user-defined-functions.md), you can get the current watermark through an API. Watermark API is supported in all our client SDKs. ### Example Golang diff --git a/docs/operations/metrics/metrics.md b/docs/operations/metrics/metrics.md index f9b030ebee..d3ca7b27d2 100644 --- a/docs/operations/metrics/metrics.md +++ b/docs/operations/metrics/metrics.md @@ -58,9 +58,9 @@ These metrics can be used to determine throughput of your pipeline. These metrics can be used to determine the latency of your pipeline. | Metric name | Metric type | Labels | Description | -| ---------------------------------------------- | ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------- | -| `source_forwarder_transformer_processing_time` | Histogram | `pipeline=`
`vertex=`
`vertex_type=`
`replica=`
`partition_name=` | Provides a histogram distribution of the processing times of User Defined Source Transformer | -| `forwarder_udf_processing_time` | Histogram | `pipeline=`
`vertex=`
`vertex_type=`
`replica=` | Provides a histogram distribution of the processing times of User Defined Functions. (UDF's) | +| ---------------------------------------------- | ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- |------------------------------------------------------------------------------------------------| +| `source_forwarder_transformer_processing_time` | Histogram | `pipeline=`
`vertex=`
`vertex_type=`
`replica=`
`partition_name=` | Provides a histogram distribution of the processing times of User-defined Source Transformer | +| `forwarder_udf_processing_time` | Histogram | `pipeline=`
`vertex=`
`vertex_type=`
`replica=` | Provides a histogram distribution of the processing times of User-defined Functions. (UDF's) | | `forwarder_forward_chunk_processing_time` | Histogram | `pipeline=`
`vertex=`
`vertex_type=`
`replica=` | Provides a histogram distribution of the processing times of the forwarder function as a whole | | `reduce_pnf_process_time` | Histogram | `pipeline=`
`vertex=`
`replica=` | Provides a histogram distribution of the processing times of the reducer | | `reduce_pnf_forward_time` | Histogram | `pipeline=`
`vertex=`
`replica=` | Provides a histogram distribution of the forwarding times of the reducer | diff --git a/docs/quick-start.md b/docs/quick-start.md index 90b0be2274..61c39d9436 100644 --- a/docs/quick-start.md +++ b/docs/quick-start.md @@ -182,7 +182,7 @@ View the UI for the advanced pipeline at https://localhost:8443/. ![Numaflow UI](assets/numaflow-ui-advanced-pipeline.png) -The source code of the `even-odd` [User Defined Function](user-guide/user-defined-functions/user-defined-functions.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd). You also can replace the [Log](./user-guide/sinks/log.md) Sink with some other sinks like [Kafka](./user-guide/sinks/kafka.md) to forward the data to Kafka topics. +The source code of the `even-odd` [user-defined function](user-guide/user-defined-functions/user-defined-functions.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd). You also can replace the [Log](./user-guide/sinks/log.md) Sink with some other sinks like [Kafka](./user-guide/sinks/kafka.md) to forward the data to Kafka topics. The pipeline can be deleted by @@ -200,6 +200,6 @@ Try more examples in the [`examples`](https://github.com/numaproj/numaflow/tree/ After exploring how Numaflow pipelines run, you can check what data [Sources](./user-guide/sources/generator.md) and [Sinks](./user-guide/sinks/kafka.md) Numaflow supports out of the box, or learn how to write -[User Defined Functions](user-guide/user-defined-functions/user-defined-functions.md). +[User-defined Functions](user-guide/user-defined-functions/user-defined-functions.md). Numaflow can also be paired with Numalogic, a collection of ML models and algorithms for real-time data analytics and AIOps including anomaly detection. Visit the [Numalogic homepage](https://numalogic.numaproj.io/) for more information. diff --git a/docs/specifications/overview.md b/docs/specifications/overview.md index 6d94b6d64e..c2c377e954 100644 --- a/docs/specifications/overview.md +++ b/docs/specifications/overview.md @@ -92,9 +92,9 @@ Logic: 1. Write to the sink; 1. Ack the data in the upstream buffer. -**UDF (User Defined Function)** +**UDF (User-defined Function)** -- User Defined Functions run in data processors. +- Use-defined Functions run in data processors. - UDFs implements a unified interface to process data. - UDFs are typically implemented by end-users, but there will be some built-in functions that can be used without writing any code. diff --git a/docs/specifications/side-inputs.md b/docs/specifications/side-inputs.md index a96b1ea5da..de1e1ba67b 100644 --- a/docs/specifications/side-inputs.md +++ b/docs/specifications/side-inputs.md @@ -1,6 +1,6 @@ # Side Inputs -Side Inputs allow the user defined functions (including UDF, UDSink, Transformer, etc.) to access slow updated data or configuration (such as database, file system, etc.) without needing to load it during each message processing. Side Inputs are read-only and can be used in both batch and streaming jobs. +Side Inputs allow the user-defined functions (including UDF, UDSink, Transformer, etc.) to access slow updated data or configuration (such as database, file system, etc.) without needing to load it during each message processing. Side Inputs are read-only and can be used in both batch and streaming jobs. ## Requirements @@ -47,8 +47,8 @@ A pipeline may have multiple Side Inputs sources, each of them will have a Side Each of the Side Inputs Manager pods contains: - An init container, which checks if the data store is ready. -- A user defined container, which runs a predefined Numaflow SDK to start a service, calling a user implemented function to get Side Input data. -- A numa container, which runs a cron like job to call the service in the user defined container, and store the returned data in the data store. +- A user-defined container, which runs a predefined Numaflow SDK to start a service, calling a user implemented function to get Side Input data. +- A numa container, which runs a cron like job to call the service in the user-defined container, and store the returned data in the data store. ![Diagram](../assets/side-inputs-manager.png) @@ -58,7 +58,7 @@ The communication protocol between the 2 containers could be based on UDS or FIF Side Inputs Manager needs to run with Active-Passive HA, which requires a leader election mechanism support. Kubernetes has a native leader election API backed by etcd, but it requires extra RBAC privileges to use it. -Considering a similar leader election mechanism is needed in some other scenarios such as Active-Passive User Defined Source, a proposal is to implement our own leader election mechanism by leveraging ISB Service. +Considering a similar leader election mechanism is needed in some other scenarios such as Active-Passive User-defined Source, a proposal is to implement our own leader election mechanism by leveraging ISB Service. #### Why NOT CronJob? @@ -70,11 +70,14 @@ Using K8s CronJob/Job will be a [challenge](https://github.com/istio/istio/issue ![Diagram](../assets/side-inputs-vertex-pod.png) -When Side Inputs is enabled for a pipeline, each of its vertex pods will have a second init container added, the init container will have a shared volume (emptyDir) mounted, and the same volume will be mounted to the User Defined Function/Sink/Transformer container. The init container reads from the data store, and saves to the shared volume. +When Side Inputs is enabled for a pipeline, each of its vertex pods will have a second init container added, +the init container will have a shared volume (emptyDir) mounted, +and the same volume will be mounted to the User-defined Function/Sink/Transformer container. +The init container reads from the data store, and saves to the shared volume. A sidecar container will also be injected by the controller, and it mounts the same volume as above. The sidecar runs a service provided by numaflow, watching the Side Inputs data from the data store, if there’s any update, reads the data and updates the shared volume. -In the User Defined Function/Sink/Sink container, a helper function will be provided by Numaflow SDK, to return the Side Input data. The helper function caches the Side Inputs data in the memory, but performs thread safe updates if it watches the changes in the shared volume. +In the User-defined Function/Sink/Sink container, a helper function will be provided by Numaflow SDK, to return the Side Input data. The helper function caches the Side Inputs data in the memory, but performs thread safe updates if it watches the changes in the shared volume. ### Numaflow SDK diff --git a/docs/user-guide/reference/side-inputs.md b/docs/user-guide/reference/side-inputs.md index 5dcaa07904..3efbe8e1da 100644 --- a/docs/user-guide/reference/side-inputs.md +++ b/docs/user-guide/reference/side-inputs.md @@ -39,7 +39,7 @@ spec: - redis ``` -### Implementing User Defined Side Inputs +### Implementing User-defined Side Inputs To use the `Side Inputs` feature, a User-defined function implementing an interface defined in the Numaflow SDK ([Go](https://github.com/numaproj/numaflow-go/blob/main/pkg/sideinput/), @@ -48,11 +48,11 @@ To use the `Side Inputs` feature, a User-defined function implementing an interf is needed to retrieve the data. You can choose the SDK of your choice to create a -User Defined Side Input image which implements the +User-defined Side Input image which implements the Side Inputs Update. #### Example in Golang -Here is an example of how to write a User Defined Side Input in Golang, +Here is an example of how to write a User-defined Side Input in Golang, ```go // handle is the side input handler function. @@ -91,7 +91,7 @@ sideinputsdk.NoBroadcastMessage() ### UDF Users need to add a watcher on the filesystem to fetch the -updated side inputs in their User Defined Source/Function/Sink +updated side inputs in their User-defined Source/Function/Sink in order to apply the new changes into the data process. For each side input there will be a file with the given path and after any update to the side input value the file will be updated. diff --git a/docs/user-guide/sinks/overview.md b/docs/user-guide/sinks/overview.md index 8007e64efa..0d22896349 100644 --- a/docs/user-guide/sinks/overview.md +++ b/docs/user-guide/sinks/overview.md @@ -14,12 +14,12 @@ Numaflow currently supports the following Sinks * [Kafka](./kafka.md) * [Log](./log.md) * [Black Hole](./blackhole.md) -* [User Defined Sink](./user-defined-sinks.md) +* [User-defined Sink](./user-defined-sinks.md) A user-defined sink is a custom Sink that a user can write using Numaflow SDK when the user needs to output the processed data to a system or using a certain transformation that is not supported by the platform's built-in sinks. As an example, once we have processed the input messages, -we can use Elasticsearch as a User defined sink to store the processed data and enable search and +we can use Elasticsearch as a user-defined sink to store the processed data and enable search and analysis on the data. ## Fallback Sink (DLQ) diff --git a/docs/user-guide/sinks/user-defined-sinks.md b/docs/user-guide/sinks/user-defined-sinks.md index 88db89abbe..830a2147bf 100644 --- a/docs/user-guide/sinks/user-defined-sinks.md +++ b/docs/user-guide/sinks/user-defined-sinks.md @@ -1,10 +1,20 @@ -# User Defined Sinks +# User-defined Sinks -A `Pipeline` may have multiple Sinks, those sinks could either be a pre-defined sink such as `kafka`, `log`, etc., or a `User Defined Sink`. +A `Pipeline` may have multiple Sinks, those sinks could either be a pre-defined sink such as `kafka`, `log`, etc., or a `user-defined sink`. -A pre-defined sink vertex runs single-container pods, a user defined sink runs two-container pods. +A pre-defined sink vertex runs single-container pods, a user-defined sink runs two-container pods. -A user defined sink vertex looks like below. +## Build Your Own User-defined Sinks + +You can build your own user-defined sinks in multiple languages. + +Check the links below to see the examples for different languages. + +- [Golang](https://github.com/numaproj/numaflow-go/tree/main/pkg/sinker/examples/) +- [Java](https://github.com/numaproj/numaflow-java/tree/main/examples/src/main/java/io/numaproj/numaflow/examples/sink/simple/) +- [Python](https://github.com/numaproj/numaflow-python/tree/main/examples/sink/) + +A user-defined sink vertex looks like below. ```yaml spec: @@ -18,10 +28,16 @@ spec: ## Available Environment Variables -Some environment variables are available in the user defined sink container: +Some environment variables are available in the user-defined sink container: - `NUMAFLOW_NAMESPACE` - Namespace. - `NUMAFLOW_POD` - Pod name. - `NUMAFLOW_REPLICA` - Replica index. - `NUMAFLOW_PIPELINE_NAME` - Name of the pipeline. - `NUMAFLOW_VERTEX_NAME` - Name of the vertex. + +## User-defined Sinks contributed from the open source community + +If you're looking for examples and usages contributed by the open source community, head over to [the numaproj-contrib repositories](https://github.com/orgs/numaproj-contrib/repositories). + +These user-defined sinks like AWS SQS, GCP Pub/Sub, provide valuable insights and guidance on how to use and write a user-defined sink. diff --git a/docs/user-guide/sources/overview.md b/docs/user-guide/sources/overview.md index 0574470265..623103cf09 100644 --- a/docs/user-guide/sources/overview.md +++ b/docs/user-guide/sources/overview.md @@ -10,8 +10,8 @@ In Numaflow, we currently support the following sources * [HTTP](./http.md) * [Ticker](./generator.md) * [Nats](./nats.md) -* [User Defined Source](./user-defined-sources.md) +* [User-defined Source](./user-defined-sources.md) -A user defined source is a custom source that a user can write using Numaflow SDK when -the user needs to read data from a system that is not supported by the platform's built-in sources. User defined source also supports custom acknowledge management for +A user-defined source is a custom source that a user can write using Numaflow SDK when +the user needs to read data from a system that is not supported by the platform's built-in sources. User-defined source also supports custom acknowledge management for exactly-once reading. diff --git a/docs/user-guide/sources/transformer/overview.md b/docs/user-guide/sources/transformer/overview.md index 3b8aa62752..c6eaf584aa 100644 --- a/docs/user-guide/sources/transformer/overview.md +++ b/docs/user-guide/sources/transformer/overview.md @@ -16,7 +16,8 @@ There are some [Built-in Transformers](builtin-transformers/README.md) that can ## Build Your Own Transformer -You can build your own transformer in multiple languages. A User Defined Transformer could be as simple as the example below in Golang. +You can build your own transformer in multiple languages. +A user-defined transformer could be as simple as the example below in Golang. In the example, the transformer extracts event times from `timestamp` of the JSON payload and assigns them to messages as new event times. It also filters out unwanted messages based on `filterOut` of the payload. ```golang diff --git a/docs/user-guide/sources/user-defined-sources.md b/docs/user-guide/sources/user-defined-sources.md index 530b4d1e2d..71d8f1dc5d 100644 --- a/docs/user-guide/sources/user-defined-sources.md +++ b/docs/user-guide/sources/user-defined-sources.md @@ -1,12 +1,12 @@ -# User Defined Sources +# User-defined Sources -A `Pipeline` may have multiple Sources, those sources could either be a pre-defined source such as `kafka`, `http`, etc., or a `User Defined Source`. +A `Pipeline` may have multiple Sources, those sources could either be a pre-defined source such as `kafka`, `http`, etc., or a `user-defined source`. With no source data transformer, A pre-defined source vertex runs single-container pods; a user-defined source runs two-container pods. -## Build Your Own User Defined Sources +## Build Your Own User-defined Sources -You can build your own user defined sources in multiple languages. +You can build your own user-defined sources in multiple languages. Check the links below to see the examples for different languages. @@ -28,10 +28,16 @@ spec: ## Available Environment Variables -Some environment variables are available in the user defined source container: +Some environment variables are available in the user-defined source container: - `NUMAFLOW_NAMESPACE` - Namespace. - `NUMAFLOW_POD` - Pod name. - `NUMAFLOW_REPLICA` - Replica index. - `NUMAFLOW_PIPELINE_NAME` - Name of the pipeline. - `NUMAFLOW_VERTEX_NAME` - Name of the vertex. + +## User-defined sources contributed from the open source community + +If you're looking for examples and usages contributed by the open source community, head over to [the numaproj-contrib repositories](https://github.com/orgs/numaproj-contrib/repositories). + +These user-defined sources like AWS SQS, GCP Pub/Sub, provide valuable insights and guidance on how to use and write a user-defined source. diff --git a/docs/user-guide/user-defined-functions/map/examples.md b/docs/user-guide/user-defined-functions/map/examples.md index 2fef3d5e0c..605f660342 100644 --- a/docs/user-guide/user-defined-functions/map/examples.md +++ b/docs/user-guide/user-defined-functions/map/examples.md @@ -112,7 +112,7 @@ View the UI for a pipeline at https://localhost:8443/. ![Numaflow UI](../../../assets/numaflow-ui-advanced-pipeline.png) -The source code of the `even-odd` [User Defined Function](../user-defined-functions.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd). You also can replace the [Log](../../sinks/log.md) Sink with some other sinks like [Kafka](../../sinks/kafka.md) to forward the data to Kafka topics. +The source code of the `even-odd` [user-defined function](../user-defined-functions.md) can be found [here](https://github.com/numaproj/numaflow-go/tree/main/pkg/mapper/examples/even_odd). You also can replace the [Log](../../sinks/log.md) Sink with some other sinks like [Kafka](../../sinks/kafka.md) to forward the data to Kafka topics. The pipeline can be deleted by diff --git a/docs/user-guide/user-defined-functions/map/map.md b/docs/user-guide/user-defined-functions/map/map.md index d4ec3901e0..2d82542d27 100644 --- a/docs/user-guide/user-defined-functions/map/map.md +++ b/docs/user-guide/user-defined-functions/map/map.md @@ -57,7 +57,7 @@ Check the links below to see the UDF examples in streaming mode for different la ### Available Environment Variables -Some environment variables are available in the user defined function container, they might be useful in your own UDF implementation. +Some environment variables are available in the user-defined function container, they might be useful in your own UDF implementation. - `NUMAFLOW_NAMESPACE` - Namespace. - `NUMAFLOW_POD` - Pod name. diff --git a/docs/user-guide/user-defined-functions/user-defined-functions.md b/docs/user-guide/user-defined-functions/user-defined-functions.md index 7dd8a92a0e..e6aef71679 100644 --- a/docs/user-guide/user-defined-functions/user-defined-functions.md +++ b/docs/user-guide/user-defined-functions/user-defined-functions.md @@ -1,6 +1,6 @@ -A `Pipeline` consists of multiple vertices, `Source`, `Sink` and `UDF(User Defined Functions)`. +A `Pipeline` consists of multiple vertices, `Source`, `Sink` and `UDF(user-defined functions)`. -User Defined Functions (UDF) is the vertex where users can run custom code to +User-defined functions (UDF) is the vertex where users can run custom code to transform the data. Data processing in the UDF is supposed to be idempotent. UDF runs as a sidecar container in a Vertex Pod, processes the received data. diff --git a/examples/README.md b/examples/README.md index d6bb098ec0..12427a5ac4 100644 --- a/examples/README.md +++ b/examples/README.md @@ -21,7 +21,7 @@ spec: rpu: 5 duration: 1s - name: cat - udf: # A user defined function + udf: # A user-defined function builtin: # Use a built-in function as the udf name: cat # cats the message - name: out diff --git a/mkdocs.yml b/mkdocs.yml index 5a5b7c15c2..43cd0a7950 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -64,9 +64,9 @@ nav: - user-guide/sinks/kafka.md - user-guide/sinks/log.md - user-guide/sinks/blackhole.md - - User Defined Sinks: "user-guide/sinks/user-defined-sinks.md" + - User-defined Sinks: "user-guide/sinks/user-defined-sinks.md" - Fallback Sink: "user-guide/sinks/fallback.md" - - User Defined Functions: + - User-defined Functions: - Overview: "user-guide/user-defined-functions/user-defined-functions.md" - Map: - Overview: "user-guide/user-defined-functions/map/map.md" diff --git a/pkg/apis/numaflow/v1alpha1/generated.proto b/pkg/apis/numaflow/v1alpha1/generated.proto index fb07daad91..d42df41e21 100644 --- a/pkg/apis/numaflow/v1alpha1/generated.proto +++ b/pkg/apis/numaflow/v1alpha1/generated.proto @@ -250,7 +250,7 @@ message CombinedEdge { optional VertexLimits toVertexLimits = 7; } -// Container is used to define the container properties for user defined functions, sinks, etc. +// Container is used to define the container properties for user-defined functions, sinks, etc. message Container { // +optional optional string image = 1; diff --git a/pkg/apis/numaflow/v1alpha1/openapi_generated.go b/pkg/apis/numaflow/v1alpha1/openapi_generated.go index 649e3ecafa..63e2ba6019 100644 --- a/pkg/apis/numaflow/v1alpha1/openapi_generated.go +++ b/pkg/apis/numaflow/v1alpha1/openapi_generated.go @@ -707,7 +707,7 @@ func schema_pkg_apis_numaflow_v1alpha1_Container(ref common.ReferenceCallback) c return common.OpenAPIDefinition{ Schema: spec.Schema{ SchemaProps: spec.SchemaProps{ - Description: "Container is used to define the container properties for user defined functions, sinks, etc.", + Description: "Container is used to define the container properties for user-defined functions, sinks, etc.", Type: []string{"object"}, Properties: map[string]spec.Schema{ "image": { diff --git a/pkg/apis/numaflow/v1alpha1/user_defined_container.go b/pkg/apis/numaflow/v1alpha1/user_defined_container.go index 03dae5c97c..369a6d3eb3 100644 --- a/pkg/apis/numaflow/v1alpha1/user_defined_container.go +++ b/pkg/apis/numaflow/v1alpha1/user_defined_container.go @@ -18,7 +18,7 @@ package v1alpha1 import corev1 "k8s.io/api/core/v1" -// Container is used to define the container properties for user defined functions, sinks, etc. +// Container is used to define the container properties for user-defined functions, sinks, etc. type Container struct { // +optional Image string `json:"image" protobuf:"bytes,1,opt,name=image"` diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go index 7fc243b09f..74663cec72 100644 --- a/pkg/apis/numaflow/v1alpha1/vertex_types.go +++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go @@ -292,7 +292,7 @@ func (v Vertex) GetPodSpec(req GetVertexPodSpecReq) (*corev1.PodSpec, error) { if containers[i].Name == CtrSideInputsWatcher { containers[i].VolumeMounts = append(containers[i].VolumeMounts, corev1.VolumeMount{Name: sideInputsVolName, MountPath: PathSideInputsMount}) } else { - // Readonly mount for user defined containers + // Readonly mount for user-defined containers containers[i].VolumeMounts = append(containers[i].VolumeMounts, corev1.VolumeMount{Name: sideInputsVolName, MountPath: PathSideInputsMount, ReadOnly: true}) } } diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go index 653bee92e5..fbc428e18c 100644 --- a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go +++ b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go @@ -288,7 +288,7 @@ func TestGetPodSpec(t *testing.T) { assert.Equal(t, CtrInit, s.InitContainers[0].Name) }) - t.Run("test user defined sink", func(t *testing.T) { + t.Run("test user-defined sink", func(t *testing.T) { testObj := testVertex.DeepCopy() testObj.Spec.Sink = &Sink{ AbstractSink: AbstractSink{ @@ -319,7 +319,7 @@ func TestGetPodSpec(t *testing.T) { assert.Contains(t, sidecarEnvNames, EnvMemoryRequest) }) - t.Run("test user defined source, with a source transformer", func(t *testing.T) { + t.Run("test user-defined source, with a source transformer", func(t *testing.T) { testObj := testVertex.DeepCopy() testObj.Spec.Source = &Source{ UDSource: &UDSource{ diff --git a/pkg/metrics/health_checker.go b/pkg/metrics/health_checker.go index 031b0eb87e..dcc7378017 100644 --- a/pkg/metrics/health_checker.go +++ b/pkg/metrics/health_checker.go @@ -18,8 +18,8 @@ package metrics import "context" -// HealthChecker is the interface to check if the user defined container is connected and ready to use +// HealthChecker is the interface to check if the user-defined container is connected and ready to use type HealthChecker interface { - // IsHealthy checks if the user defined container is healthy + // IsHealthy checks if the user-defined container is healthy IsHealthy(ctx context.Context) error } diff --git a/pkg/reconciler/vertex/controller_test.go b/pkg/reconciler/vertex/controller_test.go index d56d1db1c9..16b0cb4c50 100644 --- a/pkg/reconciler/vertex/controller_test.go +++ b/pkg/reconciler/vertex/controller_test.go @@ -301,7 +301,7 @@ func Test_BuildPodSpec(t *testing.T) { assert.Contains(t, argStr, strings.Join(testObj.GetToBuckets(), ",")) }) - t.Run("test user defined sink", func(t *testing.T) { + t.Run("test user-defined sink", func(t *testing.T) { cl := fake.NewClientBuilder().Build() r := &vertexReconciler{ client: cl, diff --git a/pkg/sinks/sink.go b/pkg/sinks/sink.go index 7439b9c54a..da4901463e 100644 --- a/pkg/sinks/sink.go +++ b/pkg/sinks/sink.go @@ -307,7 +307,7 @@ func (u *SinkProcessor) createSinkWriter(ctx context.Context, abstractSink *dfv1 } else if x := abstractSink.Blackhole; x != nil { return blackhole.NewBlackhole(ctx, u.VertexInstance) } else if x := abstractSink.UDSink; x != nil { - // if the sink is a user defined sink, then we need to pass the sinkHandler to it which will be used to invoke the user defined sink + // if the sink is a user-defined sink, then we need to pass the sinkHandler to it which will be used to invoke the user-defined sink return udsink.NewUserDefinedSink(ctx, u.VertexInstance, sinkHandler) } return nil, fmt.Errorf("invalid sink spec") diff --git a/pkg/sinks/udsink/udsink_grpc.go b/pkg/sinks/udsink/udsink_grpc.go index d52105823e..1b9c31131c 100644 --- a/pkg/sinks/udsink/udsink_grpc.go +++ b/pkg/sinks/udsink/udsink_grpc.go @@ -50,7 +50,7 @@ type SinkApplier interface { ApplySink(ctx context.Context, requests []*sinkpb.SinkRequest) []error } -// UDSgRPCBasedUDSink applies user defined sink over gRPC (over Unix Domain Socket) client/server where server is the UDSink. +// UDSgRPCBasedUDSink applies user-defined sink over gRPC (over Unix Domain Socket) client/server where server is the UDSink. type UDSgRPCBasedUDSink struct { client sinkclient.Client } @@ -105,7 +105,7 @@ func (u *UDSgRPCBasedUDSink) ApplySink(ctx context.Context, requests []*sinkpb.S } return errs } - // Use ID to map the response messages, so that there's no strict requirement for the user defined sink to return the response in order. + // Use ID to map the response messages, so that there's no strict requirement for the user-defined sink to return the response in order. resMap := make(map[string]*sinkpb.SinkResponse_Result) for _, res := range response.GetResults() { resMap[res.GetId()] = res diff --git a/pkg/sources/forward/data_forward.go b/pkg/sources/forward/data_forward.go index 41e9cc3c28..b125667439 100644 --- a/pkg/sources/forward/data_forward.go +++ b/pkg/sources/forward/data_forward.go @@ -270,9 +270,9 @@ func (df *DataForward) forwardAChunk(ctx context.Context) { messageToStep[toVertex] = make([][]isb.Message, len(df.toBuffers[toVertex])) } - // user defined transformer concurrent processing request channel + // user-defined transformer concurrent processing request channel transformerCh := make(chan *readWriteMessagePair) - // transformerResults stores the results after user defined transformer processing for all read messages. It indexes + // transformerResults stores the results after user-defined transformer processing for all read messages. It indexes // a read message to the corresponding write message transformerResults := make([]readWriteMessagePair, len(readMessages)) // applyTransformer, if there is an Internal error, it is a blocking call and diff --git a/pkg/sources/source.go b/pkg/sources/source.go index 1eb6f20091..17fc3b14c5 100644 --- a/pkg/sources/source.go +++ b/pkg/sources/source.go @@ -202,7 +202,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error { } // Readiness check if err = udsGRPCClient.WaitUntilReady(ctx); err != nil { - return fmt.Errorf("failed on user defined source readiness check, %w", err) + return fmt.Errorf("failed on user-defined source readiness check, %w", err) } defer func() { err = udsGRPCClient.CloseConn(ctx) @@ -238,7 +238,7 @@ func (sp *SourceProcessor) Start(ctx context.Context) error { // Readiness check if err = transformerGRPCClient.WaitUntilReady(ctx); err != nil { - return fmt.Errorf("failed on user defined source readiness check, %w", err) + return fmt.Errorf("failed on user-defined source readiness check, %w", err) } readyCheckers = append(readyCheckers, transformerGRPCClient) diff --git a/pkg/sources/transformer/grpc_transformer.go b/pkg/sources/transformer/grpc_transformer.go index fe15a75c49..e02d974564 100644 --- a/pkg/sources/transformer/grpc_transformer.go +++ b/pkg/sources/transformer/grpc_transformer.go @@ -33,7 +33,7 @@ import ( "github.com/numaproj/numaflow/pkg/udf/rpc" ) -// GRPCBasedTransformer applies user defined transformer over gRPC (over Unix Domain Socket) client/server where server is the transformer. +// GRPCBasedTransformer applies user-defined transformer over gRPC (over Unix Domain Socket) client/server where server is the transformer. type GRPCBasedTransformer struct { client sourcetransformer.Client } diff --git a/pkg/udf/forward/forward.go b/pkg/udf/forward/forward.go index 4a7a2e8ccb..89b91349e9 100644 --- a/pkg/udf/forward/forward.go +++ b/pkg/udf/forward/forward.go @@ -583,7 +583,7 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBufferPar needRetry := false for idx, msg := range messages { if err = errs[idx]; err != nil { - // ATM there are no user defined errors during write, all are InternalErrors. + // ATM there are no user-defined errors during write, all are InternalErrors. // Non retryable error, drop the message. Non retryable errors are only returned // when the buffer is full and the user has set the buffer full strategy to // DiscardLatest or when the message is duplicate.