Switch to using encoder in pluginmachinery (flyteorg#334)

Signed-off-by: Sean Lin <[email protected]>
Signed-off-by: Tim Bauer <[email protected]>

cmd/kubectl-flyte/cmd/printers/node.go

@@ -7,14 +7,13 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/apimachinery/pkg/util/sets"
-
 	gotree "github.com/DiSiqueira/GoTree"
 	"github.com/fatih/color"
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"github.com/flyteorg/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
 	"github.com/flyteorg/flytepropeller/pkg/controller/nodes/task"
-	"github.com/flyteorg/flytepropeller/pkg/utils"
 )
 
 var boldString = color.New(color.Bold)
@@ -59,7 +58,7 @@ func (p NodeStatusPrinter) BaseNodeInfo(node v1alpha1.BaseNode, nodeStatus v1alp
 }
 
 func (p NodeStatusPrinter) NodeInfo(wName string, node v1alpha1.BaseNode, nodeStatus v1alpha1.ExecutableNodeStatus) []string {
-	resourceName, err := utils.FixedLengthUniqueIDForParts(task.IDMaxLength, wName, node.GetID(), strconv.Itoa(int(nodeStatus.GetAttempts())))
+	resourceName, err := encoding.FixedLengthUniqueIDForParts(task.IDMaxLength, wName, node.GetID(), strconv.Itoa(int(nodeStatus.GetAttempts())))
 	if err != nil {
 		resourceName = "na"
 	}

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/benlaurie/objecthash v0.0.0-20180202135721-d1e3d6079fc1
 	github.com/fatih/color v1.10.0
 	github.com/flyteorg/flyteidl v0.21.1
-	github.com/flyteorg/flyteplugins v0.6.3
+	github.com/flyteorg/flyteplugins v0.6.5
 	github.com/flyteorg/flytestdlib v0.3.36
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-redis/redis v6.15.7+incompatible

go.sum

@@ -234,6 +234,8 @@ github.com/flyteorg/flyteidl v0.21.1 h1:wINIuKv+0xtTD0kR2RF99C5uGYuwflhY78iw+Dki
 github.com/flyteorg/flyteidl v0.21.1/go.mod h1:576W2ViEyjTpT+kEVHAGbrTP3HARNUZ/eCwrNPmdx9U=
 github.com/flyteorg/flyteplugins v0.6.3 h1:gLLQ0nISFSFnUXORxkw3QjRyHcX8AYGlHObxRT6CBk4=
 github.com/flyteorg/flyteplugins v0.6.3/go.mod h1:bEMSalcpNKAu0iFy/fTP2HJOmsCi82LcMnsB8iBeJ/U=
+github.com/flyteorg/flyteplugins v0.6.5 h1:f971kVpLEuWMbEORrLW3qq7/NhLj3jIVcewZ1/Oz9B8=
+github.com/flyteorg/flyteplugins v0.6.5/go.mod h1:bEMSalcpNKAu0iFy/fTP2HJOmsCi82LcMnsB8iBeJ/U=
 github.com/flyteorg/flytestdlib v0.3.13/go.mod h1:Tz8JCECAbX6VWGwFT6cmEQ+RJpZ/6L9pswu3fzWs220=
 github.com/flyteorg/flytestdlib v0.3.36 h1:XLvc7kfc9XkQBpPvNXevh5+Ijbgmd7gEOHTWhdEY5eA=
 github.com/flyteorg/flytestdlib v0.3.36/go.mod h1:7cDWkY3v7xsoesFcDdu6DSW5Q2U2W5KlHUbUHSwBG1Q=

pkg/controller/nodes/common/utils.go

@@ -3,9 +3,10 @@ package common
 import (
 	"strconv"
 
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
+
 	"github.com/flyteorg/flytepropeller/pkg/apis/flyteworkflow/v1alpha1"
 	"github.com/flyteorg/flytepropeller/pkg/controller/executors"
-	"github.com/flyteorg/flytepropeller/pkg/utils"
 )
 
 const maxUniqueIDLength = 20
@@ -23,7 +24,7 @@ func GenerateUniqueID(parentInfo executors.ImmutableParentInfo, nodeID string) (
 		parentRetryAttempt = strconv.Itoa(int(parentInfo.CurrentAttempt()))
 	}
 
-	return utils.FixedLengthUniqueIDForParts(maxUniqueIDLength, parentUniqueID, parentRetryAttempt, nodeID)
+	return encoding.FixedLengthUniqueIDForParts(maxUniqueIDLength, parentUniqueID, parentRetryAttempt, nodeID)
 }
 
 // When creating parentInfo, the unique id of parent is dependent on the unique id and the current attempt of the grand parent to track the lineage.

pkg/controller/nodes/dynamic/dynamic_workflow_test.go

@@ -8,10 +8,9 @@ import (
 
 	"github.com/pkg/errors"
 
-	"github.com/flyteorg/flytepropeller/pkg/utils"
-
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
 	mocks3 "github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/io/mocks"
 	"github.com/flyteorg/flytestdlib/promutils"
 	"github.com/flyteorg/flytestdlib/storage"
@@ -208,7 +207,7 @@ func Test_dynamicNodeHandler_buildContextualDynamicWorkflow_withLaunchPlans(t *t
 		assert.NotNil(t, dCtx.subWorkflowClosure)
 		assert.NotNil(t, dCtx.execContext)
 		assert.NotNil(t, dCtx.execContext.GetParentInfo())
-		expectedParentUniqueID, err := utils.FixedLengthUniqueIDForParts(20, "c1", "2", "n1")
+		expectedParentUniqueID, err := encoding.FixedLengthUniqueIDForParts(20, "c1", "2", "n1")
 		assert.Nil(t, err)
 		assert.Equal(t, expectedParentUniqueID, dCtx.execContext.GetParentInfo().GetUniqueID())
 		assert.Equal(t, uint32(1), dCtx.execContext.GetParentInfo().CurrentAttempt())
@@ -277,7 +276,7 @@ func Test_dynamicNodeHandler_buildContextualDynamicWorkflow_withLaunchPlans(t *t
 		assert.NotNil(t, dCtx.subWorkflowClosure)
 		assert.NotNil(t, dCtx.execContext)
 		assert.NotNil(t, dCtx.execContext.GetParentInfo())
-		expectedParentUniqueID, err := utils.FixedLengthUniqueIDForParts(20, "", "", "n1")
+		expectedParentUniqueID, err := encoding.FixedLengthUniqueIDForParts(20, "", "", "n1")
 		assert.Nil(t, err)
 		assert.Equal(t, expectedParentUniqueID, dCtx.execContext.GetParentInfo().GetUniqueID())
 		assert.Equal(t, uint32(1), dCtx.execContext.GetParentInfo().CurrentAttempt())
@@ -431,7 +430,7 @@ func Test_dynamicNodeHandler_buildContextualDynamicWorkflow_withLaunchPlans(t *t
 		assert.NotNil(t, dCtx.subWorkflow)
 		assert.NotNil(t, dCtx.execContext)
 		assert.NotNil(t, dCtx.execContext.GetParentInfo())
-		expectedParentUniqueID, err := utils.FixedLengthUniqueIDForParts(20, "c1", "2", "n1")
+		expectedParentUniqueID, err := encoding.FixedLengthUniqueIDForParts(20, "c1", "2", "n1")
 		assert.Nil(t, err)
 		assert.Equal(t, expectedParentUniqueID, dCtx.execContext.GetParentInfo().GetUniqueID())
 		assert.Equal(t, uint32(1), dCtx.execContext.GetParentInfo().CurrentAttempt())
@@ -576,7 +575,7 @@ func Test_dynamicNodeHandler_buildContextualDynamicWorkflow_withLaunchPlans(t *t
 		assert.NotNil(t, dCtx.subWorkflow)
 		assert.NotNil(t, dCtx.execContext)
 		assert.NotNil(t, dCtx.execContext.GetParentInfo())
-		expectedParentUniqueID, err := utils.FixedLengthUniqueIDForParts(20, "c1", "2", "n1")
+		expectedParentUniqueID, err := encoding.FixedLengthUniqueIDForParts(20, "c1", "2", "n1")
 		assert.Nil(t, err)
 		assert.Equal(t, expectedParentUniqueID, dCtx.execContext.GetParentInfo().GetUniqueID())
 		assert.Equal(t, uint32(1), dCtx.execContext.GetParentInfo().CurrentAttempt())

pkg/controller/nodes/dynamic/utils.go

@@ -7,9 +7,9 @@ import (
 
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
 
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
 	"github.com/flyteorg/flytepropeller/pkg/compiler"
 	"github.com/flyteorg/flytepropeller/pkg/controller/nodes/handler"
-	"github.com/flyteorg/flytepropeller/pkg/utils"
 )
 
 // Constructs the expected interface of a given node.
@@ -28,7 +28,7 @@ func underlyingInterface(ctx context.Context, taskReader handler.TaskReader) (*c
 }
 
 func hierarchicalNodeID(parentNodeID, retryAttempt, nodeID string) (string, error) {
-	return utils.FixedLengthUniqueIDForParts(20, parentNodeID, retryAttempt, nodeID)
+	return encoding.FixedLengthUniqueIDForParts(20, parentNodeID, retryAttempt, nodeID)
 }
 
 func updateBindingNodeIDsWithLineage(parentNodeID, retryAttempt string, binding *core.BindingData) (err error) {

pkg/controller/nodes/subworkflow/util.go

@@ -5,13 +5,13 @@ import (
 
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
 
-	"github.com/flyteorg/flytepropeller/pkg/utils"
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
 )
 
 const maxLengthForSubWorkflow = 20
 
 func GetChildWorkflowExecutionID(nodeExecID *core.NodeExecutionIdentifier, attempt uint32) (*core.WorkflowExecutionIdentifier, error) {
-	name, err := utils.FixedLengthUniqueIDForParts(maxLengthForSubWorkflow, nodeExecID.ExecutionId.Name, nodeExecID.NodeId, strconv.Itoa(int(attempt)))
+	name, err := encoding.FixedLengthUniqueIDForParts(maxLengthForSubWorkflow, nodeExecID.ExecutionId.Name, nodeExecID.NodeId, strconv.Itoa(int(attempt)))
 	if err != nil {
 		return nil, err
 	}

pkg/controller/nodes/task/taskexec_context.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"strconv"
 
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 
@@ -25,7 +26,6 @@ import (
 
 	"github.com/flyteorg/flytepropeller/pkg/controller/nodes/errors"
 	"github.com/flyteorg/flytepropeller/pkg/controller/nodes/handler"
-	"github.com/flyteorg/flytepropeller/pkg/utils"
 )
 
 var (
@@ -223,7 +223,7 @@ func (t *Handler) newTaskExecutionContext(ctx context.Context, nCtx handler.Node
 		length = *l
 	}
 
-	uniqueID, err := utils.FixedLengthUniqueIDForParts(length, nCtx.NodeExecutionMetadata().GetOwnerID().Name, currentNodeUniqueID, strconv.Itoa(int(id.RetryAttempt)))
+	uniqueID, err := encoding.FixedLengthUniqueIDForParts(length, nCtx.NodeExecutionMetadata().GetOwnerID().Name, currentNodeUniqueID, strconv.Itoa(int(id.RetryAttempt)))
 	if err != nil {
 		// SHOULD never really happen
 		return nil, err

pkg/utils/secrets/marshaler.go

@@ -5,7 +5,7 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/flyteorg/flytepropeller/pkg/utils"
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
 
 	"github.com/golang/protobuf/proto"
 
@@ -23,12 +23,12 @@ const (
 const totalAnnotationSizeLimitB int = 256 * (1 << 10) // 256 kB
 
 func encodeSecret(secretAsString string) string {
-	res := utils.Base32Encoder.EncodeToString([]byte(secretAsString))
+	res := encoding.Base32Encoder.EncodeToString([]byte(secretAsString))
 	return strings.TrimSuffix(res, "=")
 }
 
 func decodeSecret(encoded string) (string, error) {
-	decodedRaw, err := utils.Base32Encoder.DecodeString(encoded)
+	decodedRaw, err := encoding.Base32Encoder.DecodeString(encoded)
 	if err != nil {
 		return encoded, err
 	}

pkg/webhook/utils.go

@@ -5,7 +5,7 @@ import (
 	"path/filepath"
 	"strings"
 
-	"github.com/flyteorg/flytepropeller/pkg/utils"
+	"github.com/flyteorg/flyteplugins/go/tasks/pluginmachinery/encoding"
 
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
 
@@ -40,7 +40,7 @@ func CreateEnvVarForSecret(secret *core.Secret) corev1.EnvVar {
 func CreateVolumeForSecret(secret *core.Secret) corev1.Volume {
 	return corev1.Volume{
 		// we don't want to create different volume for the same secret group
-		Name: utils.Base32Encoder.EncodeToString([]byte(secret.Group + EnvVarGroupKeySeparator + secret.GroupVersion)),
+		Name: encoding.Base32Encoder.EncodeToString([]byte(secret.Group + EnvVarGroupKeySeparator + secret.GroupVersion)),
 		VolumeSource: corev1.VolumeSource{
 			Secret: &corev1.SecretVolumeSource{
 				SecretName: secret.Group,