Repository: numaproj/numaflow
Pull request #2087 (2024-09-25 15:18:20)

feat: container-type level version compatibility check
introduces one more dimension when checking version compatibly - container type. 

<!--

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.

-->


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/numaproj/numaflow/pull/2087?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj) Report
Attention: Patch coverage is `37.50000%` with `10 lines` in your changes missing coverage. Please review.
> Project coverage is 64.36%. Comparing base [(`6d1ebd0`)](https://app.codecov.io/gh/numaproj/numaflow/commit/6d1ebd04f2089c81bd8e0c5e763cd7c363cb7623?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj) to head [(`88e3ddf`)](https://app.codecov.io/gh/numaproj/numaflow/commit/88e3ddf82b5042c7c4cd9f3f7ddb13bc01640043?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj).

| [Files with missing lines](https://app.codecov.io/gh/numaproj/numaflow/pull/2087?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj) | Patch % | Lines |
|---|---|---|
| [pkg/sdkclient/serverinfo/serverinfo.go](https://app.codecov.io/gh/numaproj/numaflow/pull/2087?src=pr&el=tree&filepath=pkg%2Fsdkclient%2Fserverinfo%2Fserverinfo.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj#diff-cGtnL3Nka2NsaWVudC9zZXJ2ZXJpbmZvL3NlcnZlcmluZm8uZ28=) | 37.50% | [5 Missing and 5 partials :warning: ](https://app.codecov.io/gh/numaproj/numaflow/pull/2087?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #2087      +/-   ##
==========================================
- Coverage   64.47%   64.36%   -0.12%     
==========================================
  Files         322      322              
  Lines       30171    30184      +13     
==========================================
- Hits        19454    19428      -26     
- Misses       9692     9727      +35     
- Partials     1025     1029       +4     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/numaproj/numaflow/pull/2087?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj).


Patch:
diff --git a/pkg/sdkclient/const.go b/pkg/sdkclient/const.go
index b7ace077..54f88b66 100644
--- a/pkg/sdkclient/const.go
+++ b/pkg/sdkclient/const.go
@@ -38,7 +38,6 @@ const (
 
 	// Server information file configs
 	MapServerInfoFile               = "/var/run/numaflow/mapper-server-info"
-	MapStreamServerInfoFile         = "/var/run/numaflow/mapstreamer-server-info"
 	ReduceServerInfoFile            = "/var/run/numaflow/reducer-server-info"
 	ReduceStreamServerInfoFile      = "/var/run/numaflow/reducestreamer-server-info"
 	SessionReduceServerInfoFile     = "/var/run/numaflow/sessionreducer-server-info"
@@ -47,5 +46,4 @@ const (
 	FbSinkServerInfoFile            = "/var/run/numaflow/fb-sinker-server-info"
 	SourceServerInfoFile            = "/var/run/numaflow/sourcer-server-info"
 	SourceTransformerServerInfoFile = "/var/run/numaflow/sourcetransformer-server-info"
-	BatchMapServerInfoFile          = "/var/run/numaflow/batchmapper-server-info"
 )
diff --git a/pkg/sdkclient/serverinfo/serverinfo.go b/pkg/sdkclient/serverinfo/serverinfo.go
index d01acd1c..f94d83e0 100644
--- a/pkg/sdkclient/serverinfo/serverinfo.go
+++ b/pkg/sdkclient/serverinfo/serverinfo.go
@@ -67,6 +67,10 @@ func waitForServerInfo(timeout time.Duration, filePath string) (*ServerInfo, err
 	minNumaflowVersion := serverInfo.MinimumNumaflowVersion
 	sdkLanguage := serverInfo.Language
 	numaflowVersion := numaflow.GetVersion().Version
+	containerType, err := getContainerType(filePath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get container type: %w", err)
+	}
 
 	// If MinimumNumaflowVersion is empty, skip the numaflow compatibility check as there was an
 	// error writing server info file on the SDK side
@@ -87,7 +91,7 @@ func waitForServerInfo(timeout time.Duration, filePath string) (*ServerInfo, err
 	if sdkVersion == "" || sdkLanguage == "" {
 		log.Printf("warning: failed to get the SDK version/language, skipping SDK version compatibility check")
 	} else {
-		if err := checkSDKCompatibility(sdkVersion, sdkLanguage, minimumSupportedSDKVersions); err != nil {
+		if err := checkSDKCompatibility(sdkVersion, sdkLanguage, containerType, minimumSupportedSDKVersions); err != nil {
 			return nil, fmt.Errorf("SDK version %s does not satisfy the minimum required by numaflow version %s: %w",
 				sdkVersion, numaflowVersion, err)
 		}
@@ -176,8 +180,11 @@ func checkNumaflowCompatibility(numaflowVersion string, minNumaflowVersion strin
 }
 
 // checkSDKCompatibility checks if the current SDK version is compatible with the numaflow version
-func checkSDKCompatibility(sdkVersion string, sdkLanguage Language, minSupportedSDKVersions sdkConstraints) error {
-	if sdkRequiredVersion, ok := minSupportedSDKVersions[sdkLanguage]; ok {
+func checkSDKCompatibility(sdkVersion string, sdkLanguage Language, containerType ContainerType, minSupportedSDKVersions sdkConstraints) error {
+	if _, ok := minSupportedSDKVersions[sdkLanguage]; !ok {
+		return fmt.Errorf("SDK language %s is not supported", sdkLanguage)
+	}
+	if sdkRequiredVersion, ok := minSupportedSDKVersions[sdkLanguage][containerType]; ok {
 		sdkConstraint := fmt.Sprintf(">= %s", sdkRequiredVersion)
 		if sdkLanguage == Python {
 			// Python pre-releases/releases follow PEP440 specification which requires a different library for parsing
@@ -206,6 +213,19 @@ func checkSDKCompatibility(sdkVersion string, sdkLanguage Language, minSupported
 					sdkVersionSemVer.String(), humanReadable(sdkRequiredVersion), err)
 			}
 		}
+	} else {
+		return fmt.Errorf("SDK container type %s is not supported", containerType)
 	}
 	return nil
 }
+
+// getContainerType returns the container type from the server info file path
+// serverInfoFilePath is in the format of "/var/run/numaflow/{ContainerType}-server-info"
+func getContainerType(serverInfoFilePath string) (ContainerType, error) {
+	splits := strings.Split(serverInfoFilePath, "/")
+	if containerType := strings.TrimSuffix(splits[len(splits)-1], "-server-info"); containerType == "" {
+		return "", fmt.Errorf("failed to get container type from server info file path: %s", serverInfoFilePath)
+	} else {
+		return ContainerType(containerType), nil
+	}
+}
diff --git a/pkg/sdkclient/serverinfo/types.go b/pkg/sdkclient/serverinfo/types.go
index 9e4a152d..23f6e239 100644
--- a/pkg/sdkclient/serverinfo/types.go
+++ b/pkg/sdkclient/serverinfo/types.go
@@ -27,7 +27,23 @@ const (
 	Rust   Language = "rust"
 )
 
-type sdkConstraints map[Language]string
+type ContainerType string
+
+// the string content matches the corresponding server info file name.
+// DO NOT change it unless the server info file name is changed.
+const (
+	sourcer           ContainerType = "sourcer"
+	sourcetransformer ContainerType = "sourcetransformer"
+	sinker            ContainerType = "sinker"
+	mapper            ContainerType = "mapper"
+	reducer           ContainerType = "reducer"
+	reducestreamer    ContainerType = "reducestreamer"
+	sessionreducer    ContainerType = "sessionreducer"
+	sideinput         ContainerType = "sideinput"
+	fbsinker          ContainerType = "fb-sinker"
+)
+
+type sdkConstraints map[Language]map[ContainerType]string
 
 /*
 minimumSupportedSDKVersions is the minimum supported version of each SDK for the current numaflow version.
@@ -70,14 +86,54 @@ A constraint ">=0.8.0-z" will match any pre-release version of 0.8.0, including
 More details about version comparison can be found in the PEP 440 and semver documentation.
 */
 var minimumSupportedSDKVersions = sdkConstraints{
-	// meaning the minimum supported python SDK version is 0.8.0
-	Python: "0.8.0rc100",
-	// meaning the minimum supported go SDK version is 0.8.0
-	Go: "0.8.0-z",
-	// meaning the minimum supported java SDK version is 0.8.0
-	Java: "0.8.0-z",
-	// meaning the minimum supported rust SDK version is 0.1.0
-	Rust: "0.1.0-z",
+	Python: map[ContainerType]string{
+		// meaning the minimum supported python SDK version is 0.8.0
+		sourcer:           "0.8.0rc100",
+		sourcetransformer: "0.8.0rc100",
+		sinker:            "0.8.0rc100",
+		mapper:            "0.8.0rc100",
+		reducer:           "0.8.0rc100",
+		reducestreamer:    "0.8.0rc100",
+		sessionreducer:    "0.8.0rc100",
+		sideinput:         "0.8.0rc100",
+		fbsinker:          "0.8.0rc100",
+	},
+	Go: map[ContainerType]string{
+		// meaning the minimum supported go SDK version is 0.8.0
+		sourcer:           "0.8.0-z",
+		sourcetransformer: "0.8.0-z",
+		sinker:            "0.8.0-z",
+		mapper:            "0.8.0-z",
+		reducer:           "0.8.0-z",
+		reducestreamer:    "0.8.0-z",
+		sessionreducer:    "0.8.0-z",
+		sideinput:         "0.8.0-z",
+		fbsinker:          "0.8.0-z",
+	},
+	Java: map[ContainerType]string{
+		// meaning the minimum supported go SDK version is 0.8.0
+		sourcer:           "0.8.0-z",
+		sourcetransformer: "0.8.0-z",
+		sinker:            "0.8.0-z",
+		mapper:            "0.8.0-z",
+		reducer:           "0.8.0-z",
+		reducestreamer:    "0.8.0-z",
+		sessionreducer:    "0.8.0-z",
+		sideinput:         "0.8.0-z",
+		fbsinker:          "0.8.0-z",
+	},
+	Rust: map[ContainerType]string{
+		// meaning the minimum supported go SDK version is 0.1.0
+		sourcer:           "0.1.0-z",
+		sourcetransformer: "0.1.0-z",
+		sinker:            "0.1.0-z",
+		mapper:            "0.1.0-z",
+		reducer:           "0.1.0-z",
+		reducestreamer:    "0.1.0-z",
+		sessionreducer:    "0.1.0-z",
+		sideinput:         "0.1.0-z",
+		fbsinker:          "0.1.0-z",
+	},
 }
 
 // humanReadable returns the human-readable minimum supported version.


Test patch:
diff --git a/pkg/sdkclient/serverinfo/serverinfo_test.go b/pkg/sdkclient/serverinfo/serverinfo_test.go
index ad7f06e6..105775de 100644
--- a/pkg/sdkclient/serverinfo/serverinfo_test.go
+++ b/pkg/sdkclient/serverinfo/serverinfo_test.go
@@ -29,7 +29,7 @@ import (
 )
 
 func Test_SDKServerInfo(t *testing.T) {
-	filepath := os.TempDir() + "/server-info"
+	filepath := os.TempDir() + "/sourcer-server-info"
 	defer os.Remove(filepath)
 	info := &ServerInfo{
 		Protocol:               TCP,
@@ -185,10 +185,18 @@ func Test_CheckNumaflowCompatibility(t *testing.T) {
 // this test suite is to test SDK compatibility check when all the minimum-supported versions are stable releases
 func Test_CheckSDKCompatibility_MinimumBeingStableReleases(t *testing.T) {
 	var testMinimumSupportedSDKVersions = sdkConstraints{
-		Python: "0.6.0rc100",
-		Go:     "0.6.0-z",
-		Java:   "0.6.0-z",
-		Rust:   "0.1.0-z",
+		Python: map[ContainerType]string{
+			sourcer: "0.6.0rc100",
+		},
+		Go: map[ContainerType]string{
+			sourcer: "0.6.0-z",
+		},
+		Java: map[ContainerType]string{
+			sourcer: "0.6.0-z",
+		},
+		Rust: map[ContainerType]string{
+			sourcer: "0.1.0-z",
+		},
 	}
 	tests := []struct {
 		name                        string
@@ -275,7 +283,7 @@ func Test_CheckSDKCompatibility_MinimumBeingStableReleases(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err := checkSDKCompatibility(tt.sdkVersion, tt.sdkLanguage, tt.minimumSupportedSDKVersions)
+			err := checkSDKCompatibility(tt.sdkVersion, tt.sdkLanguage, sourcer, tt.minimumSupportedSDKVersions)
 			if tt.shouldErr {
 				assert.Error(t, err, "Expected error")
 				assert.Contains(t, err.Error(), tt.errMessage)
@@ -289,10 +297,18 @@ func Test_CheckSDKCompatibility_MinimumBeingStableReleases(t *testing.T) {
 // this test suite is to test SDK compatibility check when all the minimum-supported versions are pre-releases
 func Test_CheckSDKCompatibility_MinimumBeingPreReleases(t *testing.T) {
 	var testMinimumSupportedSDKVersions = sdkConstraints{
-		Python: "0.6.0b1",
-		Go:     "0.6.0-rc2",
-		Java:   "0.6.0-rc2",
-		Rust:   "0.1.0-rc3",
+		Python: map[ContainerType]string{
+			sourcer: "0.6.0b1",
+		},
+		Go: map[ContainerType]string{
+			sourcer: "0.6.0-rc2",
+		},
+		Java: map[ContainerType]string{
+			sourcer: "0.6.0-rc2",
+		},
+		Rust: map[ContainerType]string{
+			sourcer: "0.1.0-rc3",
+		},
 	}
 	tests := []struct {
 		name                        string
@@ -379,7 +395,7 @@ func Test_CheckSDKCompatibility_MinimumBeingPreReleases(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err := checkSDKCompatibility(tt.sdkVersion, tt.sdkLanguage, tt.minimumSupportedSDKVersions)
+			err := checkSDKCompatibility(tt.sdkVersion, tt.sdkLanguage, sourcer, tt.minimumSupportedSDKVersions)
 			if tt.shouldErr {
 				assert.Error(t, err, "Expected error")
 				assert.Contains(t, err.Error(), tt.errMessage)