Repository: numaproj/numaflow
Pull request #812 (2023-06-28 17:02:44)

chore: some code refactor and docs change
Explain what this PR does.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->



Patch:
diff --git a/docs/user-guide/reference/configuration/access-path.md b/docs/operations/ui-access-path.md
similarity index 100%
rename from docs/user-guide/reference/configuration/access-path.md
rename to docs/operations/ui-access-path.md
diff --git a/docs/user-guide/reference/kustomize/kustomize.md b/docs/user-guide/reference/kustomize/kustomize.md
index 732a775e..6ec38a4a 100644
--- a/docs/user-guide/reference/kustomize/kustomize.md
+++ b/docs/user-guide/reference/kustomize/kustomize.md
@@ -11,10 +11,10 @@ apiVersion: kustomize.config.k8s.io/v1beta1
 configurations:
   - numaflow-transformer-config.yaml
   # Or reference the remote configuration directly.
-  # - https://raw.githubusercontent.com/numaproj/numaflow/main/docs/user-guide/kustomize/numaflow-transformer-config.yaml
+  # - https://raw.githubusercontent.com/numaproj/numaflow/main/docs/user-guide/reference/kustomize/numaflow-transformer-config.yaml
 ```
 
-Here is an [example](https://github.com/numaproj/numaflow/blob/main/docs/user-guide/kustomize/examples/transformer) to use transformers with a Pipeline.
+Here is an [example](https://github.com/numaproj/numaflow/tree/main/docs/user-guide/reference/kustomize/examples/transformer) to use transformers with a Pipeline.
 
 ## Patch
 
@@ -84,4 +84,4 @@ patchesStrategicMerge:
               rpu: 500
 ```
 
-See the full example [here](https://github.com/numaproj/numaflow/blob/main/docs/user-guide/kustomize/examples/patch).
+See the full example [here](https://github.com/numaproj/numaflow/tree/main/docs/user-guide/reference/kustomize/examples/patch).
diff --git a/mkdocs.yml b/mkdocs.yml
index d7f1403d..e1f39d94 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -58,6 +58,7 @@ nav:
                   - Overview: "user-guide/sources/transformer/builtin-transformers/README.md"
                   - Filter: "user-guide/sources/transformer/builtin-transformers/filter.md"
                   - Event Time Extractor: "user-guide/sources/transformer/builtin-transformers/event-time-extractor.md"
+                  - Event Time Extraction Filter: "user-guide/sources/transformer/builtin-transformers/time-extraction-filter.md"
       - Sinks:
           - Overview: "user-guide/sinks/overview.md"
           - user-guide/sinks/kafka.md
@@ -91,7 +92,6 @@ nav:
             - user-guide/reference/configuration/init-containers.md
             - user-guide/reference/configuration/sidecar-containers.md
             - user-guide/reference/configuration/pipeline-customization.md
-            - user-guide/reference/configuration/access-path.md
           - user-guide/reference/kustomize/kustomize.md
           - APIs.md
   - Operator Manual:
@@ -99,6 +99,7 @@ nav:
       - operations/installation.md
       - Configuration:
           - Controller Configuration: "operations/controller-configmap.md"
+          - UI Server Access Path: "operations/ui-access-path.md"
           - operations/metrics/metrics.md
           - operations/grafana.md
   - Contributor Guide:
diff --git a/pkg/udferr/error.go b/pkg/sdkclient/error/error.go
similarity index 98%
rename from pkg/udferr/error.go
rename to pkg/sdkclient/error/error.go
index b435db99..54b6e8f6 100644
--- a/pkg/udferr/error.go
+++ b/pkg/sdkclient/error/error.go
@@ -1,4 +1,4 @@
-package udferr
+package error
 
 import (
 	"fmt"
diff --git a/pkg/sdkclient/udf/client/client.go b/pkg/sdkclient/udf/client/client.go
index ab860bf6..1833c203 100644
--- a/pkg/sdkclient/udf/client/client.go
+++ b/pkg/sdkclient/udf/client/client.go
@@ -3,7 +3,6 @@ package client
 import (
 	"context"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/udferr"
 	"io"
 	"log"
 	"strconv"
@@ -20,6 +19,7 @@ import (
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
 	"github.com/numaproj/numaflow-go/pkg/function"
 	"github.com/numaproj/numaflow-go/pkg/info"
+	sdkerr "github.com/numaproj/numaflow/pkg/sdkclient/error"
 )
 
 // client contains the grpc connection and the grpc client.
@@ -217,7 +217,7 @@ func toUDFErr(name string, err error) error {
 	}
 	statusCode, ok := status.FromError(err)
 	// default udfError
-	udfError := udferr.New(udferr.NonRetryable, statusCode.Message())
+	udfError := sdkerr.New(sdkerr.NonRetryable, statusCode.Message())
 	// check if it's a standard status code
 	if !ok {
 		// if not, the status code will be unknown which we consider as non retryable
@@ -230,7 +230,7 @@ func toUDFErr(name string, err error) error {
 		return nil
 	case codes.DeadlineExceeded, codes.Unavailable, codes.Unknown:
 		// update to retryable err
-		udfError = udferr.New(udferr.Retryable, statusCode.Message())
+		udfError = sdkerr.New(sdkerr.Retryable, statusCode.Message())
 		log.Printf("failed %s: %s", name, udfError.Error())
 		return udfError
 	default:
diff --git a/pkg/sources/transformer/grpc_transformer.go b/pkg/sources/transformer/grpc_transformer.go
index 1a01be8c..08594c7c 100644
--- a/pkg/sources/transformer/grpc_transformer.go
+++ b/pkg/sources/transformer/grpc_transformer.go
@@ -19,18 +19,18 @@ package transformer
 import (
 	"context"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 	"time"
 
+	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+	"k8s.io/apimachinery/pkg/util/wait"
+
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
 	"github.com/numaproj/numaflow/pkg/forward/applier"
 	"github.com/numaproj/numaflow/pkg/isb"
+	sdkerr "github.com/numaproj/numaflow/pkg/sdkclient/error"
+	"github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 	"github.com/numaproj/numaflow/pkg/udf/function"
-	"github.com/numaproj/numaflow/pkg/udferr"
-	"k8s.io/apimachinery/pkg/util/wait"
-
-	"google.golang.org/protobuf/types/known/emptypb"
-	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 // gRPCBasedTransformer applies user defined transformer over gRPC (over Unix Domain Socket) client/server where server is the transformer.
@@ -93,9 +93,9 @@ func (u *gRPCBasedTransformer) ApplyMap(ctx context.Context, readMessage *isb.Re
 
 	datumList, err := u.client.MapTFn(ctx, d)
 	if err != nil {
-		udfErr, _ := udferr.FromError(err)
+		udfErr, _ := sdkerr.FromError(err)
 		switch udfErr.ErrorKind() {
-		case udferr.Retryable:
+		case sdkerr.Retryable:
 			var success bool
 			_ = wait.ExponentialBackoffWithContext(ctx, wait.Backoff{
 				// retry every "duration * factor + [0, jitter]" interval for 5 times
@@ -106,11 +106,11 @@ func (u *gRPCBasedTransformer) ApplyMap(ctx context.Context, readMessage *isb.Re
 			}, func() (done bool, err error) {
 				datumList, err = u.client.MapTFn(ctx, d)
 				if err != nil {
-					udfErr, _ = udferr.FromError(err)
+					udfErr, _ = sdkerr.FromError(err)
 					switch udfErr.ErrorKind() {
-					case udferr.Retryable:
+					case sdkerr.Retryable:
 						return false, nil
-					case udferr.NonRetryable:
+					case sdkerr.NonRetryable:
 						return true, nil
 					default:
 						return true, nil
@@ -129,7 +129,7 @@ func (u *gRPCBasedTransformer) ApplyMap(ctx context.Context, readMessage *isb.Re
 					},
 				}
 			}
-		case udferr.NonRetryable:
+		case sdkerr.NonRetryable:
 			return nil, function.ApplyUDFErr{
 				UserUDFErr: false,
 				Message:    fmt.Sprintf("gRPC client.MapFn failed, %s", err),
diff --git a/pkg/udf/function/uds_grpc.go b/pkg/udf/function/uds_grpc.go
index 45fb6aff..61c3f15f 100644
--- a/pkg/udf/function/uds_grpc.go
+++ b/pkg/udf/function/uds_grpc.go
@@ -19,24 +19,24 @@ package function
 import (
 	"context"
 	"fmt"
-	clientsdk "github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 	"strconv"
 	"sync"
 	"time"
 
-	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
-	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
-	"github.com/numaproj/numaflow/pkg/udferr"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"k8s.io/apimachinery/pkg/util/wait"
 
+	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
+	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 	map_applier "github.com/numaproj/numaflow/pkg/forward/applier"
 	"github.com/numaproj/numaflow/pkg/isb"
 	reduce_applier "github.com/numaproj/numaflow/pkg/reduce/applier"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	sdkerr "github.com/numaproj/numaflow/pkg/sdkclient/error"
+	clientsdk "github.com/numaproj/numaflow/pkg/sdkclient/udf/client"
 )
 
 // UDSgRPCBasedUDF applies user defined function over gRPC (over Unix Domain Socket) client/server where server is the UDF.
@@ -101,9 +101,9 @@ func (u *UDSgRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMes
 
 	datumList, err := u.client.MapFn(ctx, d)
 	if err != nil {
-		udfErr, _ := udferr.FromError(err)
+		udfErr, _ := sdkerr.FromError(err)
 		switch udfErr.ErrorKind() {
-		case udferr.Retryable:
+		case sdkerr.Retryable:
 			var success bool
 			_ = wait.ExponentialBackoffWithContext(ctx, wait.Backoff{
 				// retry every "duration * factor + [0, jitter]" interval for 5 times
@@ -114,11 +114,11 @@ func (u *UDSgRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMes
 			}, func() (done bool, err error) {
 				datumList, err = u.client.MapFn(ctx, d)
 				if err != nil {
-					udfErr, _ = udferr.FromError(err)
+					udfErr, _ = sdkerr.FromError(err)
 					switch udfErr.ErrorKind() {
-					case udferr.Retryable:
+					case sdkerr.Retryable:
 						return false, nil
-					case udferr.NonRetryable:
+					case sdkerr.NonRetryable:
 						return true, nil
 					default:
 						return true, nil
@@ -137,7 +137,7 @@ func (u *UDSgRPCBasedUDF) ApplyMap(ctx context.Context, readMessage *isb.ReadMes
 					},
 				}
 			}
-		case udferr.NonRetryable:
+		case sdkerr.NonRetryable:
 			return nil, ApplyUDFErr{
 				UserUDFErr: false,
 				Message:    fmt.Sprintf("gRPC client.MapFn failed, %s", err),
@@ -294,9 +294,9 @@ readLoop:
 	if err != nil {
 		// if any error happens in reduce
 		// will exit and restart the numa container
-		udfErr, _ := udferr.FromError(err)
+		udfErr, _ := sdkerr.FromError(err)
 		switch udfErr.ErrorKind() {
-		case udferr.Retryable:
+		case sdkerr.Retryable:
 			// TODO: currently we don't handle retryable errors for reduce
 			return nil, ApplyUDFErr{
 				UserUDFErr: false,
@@ -306,7 +306,7 @@ readLoop:
 					MainCarDown: false,
 				},
 			}
-		case udferr.NonRetryable:
+		case sdkerr.NonRetryable:
 			return nil, ApplyUDFErr{
 				UserUDFErr: false,
 				Message:    fmt.Sprintf("gRPC client.ReduceFn failed, %s", err),


Test patch:
diff --git a/pkg/udferr/error_test.go b/pkg/sdkclient/error/error_test.go
similarity index 99%
rename from pkg/udferr/error_test.go
rename to pkg/sdkclient/error/error_test.go
index 34eca429..9c22ee13 100644
--- a/pkg/udferr/error_test.go
+++ b/pkg/sdkclient/error/error_test.go
@@ -1,4 +1,4 @@
-package udferr
+package error
 
 import (
 	"fmt"
diff --git a/pkg/sdkclient/udf/clienttest/clienttest.go b/pkg/sdkclient/udf/clienttest/clienttest.go
index acf0f09e..993c1ca1 100644
--- a/pkg/sdkclient/udf/clienttest/clienttest.go
+++ b/pkg/sdkclient/udf/clienttest/clienttest.go
@@ -3,7 +3,7 @@ package clienttest
 import (
 	"context"
 	"fmt"
-	"github.com/numaproj/numaflow/pkg/udferr"
+
 	"io"
 	"log"
 
@@ -14,6 +14,7 @@ import (
 
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
 	"github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1/funcmock"
+	sdkerr "github.com/numaproj/numaflow/pkg/sdkclient/error"
 )
 
 // client contains the grpc client for testing.
@@ -151,7 +152,7 @@ func toUDFErr(name string, err error) error {
 	}
 	statusCode, ok := status.FromError(err)
 	// default udfError
-	udfError := udferr.New(udferr.NonRetryable, statusCode.Message())
+	udfError := sdkerr.New(sdkerr.NonRetryable, statusCode.Message())
 	// check if it's a standard status code
 	if !ok {
 		// if not, the status code will be unknown which we consider as non retryable
@@ -164,7 +165,7 @@ func toUDFErr(name string, err error) error {
 		return nil
 	case codes.DeadlineExceeded, codes.Unavailable, codes.Unknown:
 		// update to retryable err
-		udfError = udferr.New(udferr.Retryable, statusCode.Message())
+		udfError = sdkerr.New(sdkerr.Retryable, statusCode.Message())
 		log.Printf("failed %s: %s", name, udfError.Error())
 		return udfError
 	default: