chore: Deprecate functional library in favor of lo and operatorpkg (k…

…ubernetes-sigs#1398)

go.mod

@@ -5,7 +5,7 @@ go 1.22.5
 require (
 	github.com/Pallinder/go-randomdata v1.2.0
 	github.com/avast/retry-go v3.0.0+incompatible
-	github.com/awslabs/operatorpkg v0.0.0-20240628210115-2457d6af0d2f
+	github.com/awslabs/operatorpkg v0.0.0-20240701195752-116cbcffbcb4
 	github.com/docker/docker v27.0.2+incompatible
 	github.com/go-logr/logr v1.4.2
 	github.com/imdario/mergo v0.3.16

go.sum

@@ -47,8 +47,8 @@ github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk5
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/avast/retry-go v3.0.0+incompatible h1:4SOWQ7Qs+oroOTQOYnAHqelpCO0biHSxpiH9JdtuBj0=
 github.com/avast/retry-go v3.0.0+incompatible/go.mod h1:XtSnn+n/sHqQIpZ10K1qAevBhOOCWBLXXy3hyiqqBrY=
-github.com/awslabs/operatorpkg v0.0.0-20240628210115-2457d6af0d2f h1:QSA8dxtEmwMbObJjF7FkLfTz21qRis0zLMAAnS7aTNA=
-github.com/awslabs/operatorpkg v0.0.0-20240628210115-2457d6af0d2f/go.mod h1:RxolNq1josGwaVEB5I19Y7dX01MPJdDppaM6tI8R4Q0=
+github.com/awslabs/operatorpkg v0.0.0-20240701195752-116cbcffbcb4 h1:mD24yp98VHBV3PympU2jTKAzKq1IIgpdZd9+aJOuxv8=
+github.com/awslabs/operatorpkg v0.0.0-20240701195752-116cbcffbcb4/go.mod h1:oQUBEhsmTceyAkUb7XRIYsMKvJkidoU3UpAa+beK/0w=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=

kwok/apis/apis.go

@@ -19,10 +19,8 @@ package apis
 import (
 	_ "embed"
 
-	"github.com/samber/lo"
+	"github.com/awslabs/operatorpkg/object"
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
-
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 const (
@@ -34,6 +32,6 @@ var (
 	//go:embed crds/karpenter.kwok.sh_kwoknodeclasses.yaml
 	KWOKNodeClassCRD []byte
 	CRDs             = []*v1.CustomResourceDefinition{
-		lo.Must(functional.Unmarshal[v1.CustomResourceDefinition](KWOKNodeClassCRD)),
+		object.Unmarshal[v1.CustomResourceDefinition](KWOKNodeClassCRD),
 	}
 )

pkg/apis/apis.go

@@ -19,10 +19,8 @@ package apis
 import (
 	_ "embed"
 
-	"github.com/samber/lo"
+	"github.com/awslabs/operatorpkg/object"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
-
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 const (
@@ -37,7 +35,7 @@ var (
 	//go:embed crds/karpenter.sh_nodeclaims.yaml
 	NodeClaimCRD []byte
 	CRDs         = []*apiextensionsv1.CustomResourceDefinition{
-		lo.Must(functional.Unmarshal[apiextensionsv1.CustomResourceDefinition](NodePoolCRD)),
-		lo.Must(functional.Unmarshal[apiextensionsv1.CustomResourceDefinition](NodeClaimCRD)),
+		object.Unmarshal[apiextensionsv1.CustomResourceDefinition](NodePoolCRD),
+		object.Unmarshal[apiextensionsv1.CustomResourceDefinition](NodeClaimCRD),
 	}
 )

pkg/cloudprovider/fake/cloudprovider.go

@@ -38,7 +38,6 @@ import (
 	"sigs.k8s.io/karpenter/pkg/cloudprovider"
 	"sigs.k8s.io/karpenter/pkg/scheduling"
 	"sigs.k8s.io/karpenter/pkg/test"
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 	"sigs.k8s.io/karpenter/pkg/utils/resources"
 )
 
@@ -150,8 +149,8 @@ func (c *CloudProvider) Create(ctx context.Context, nodeClaim *v1beta1.NodeClaim
 		Spec: *nodeClaim.Spec.DeepCopy(),
 		Status: v1beta1.NodeClaimStatus{
 			ProviderID:  test.RandomProviderID(),
-			Capacity:    functional.FilterMap(instanceType.Capacity, func(_ v1.ResourceName, v resource.Quantity) bool { return !resources.IsZero(v) }),
-			Allocatable: functional.FilterMap(instanceType.Allocatable(), func(_ v1.ResourceName, v resource.Quantity) bool { return !resources.IsZero(v) }),
+			Capacity:    lo.PickBy(instanceType.Capacity, func(_ v1.ResourceName, v resource.Quantity) bool { return !resources.IsZero(v) }),
+			Allocatable: lo.PickBy(instanceType.Allocatable(), func(_ v1.ResourceName, v resource.Quantity) bool { return !resources.IsZero(v) }),
 		},
 	}
 	c.CreatedNodeClaims[created.Status.ProviderID] = created

pkg/controllers/provisioning/provisioner.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/awslabs/operatorpkg/option"
 	"github.com/awslabs/operatorpkg/status"
 
 	"github.com/awslabs/operatorpkg/singleton"
@@ -48,7 +49,6 @@ import (
 
 	"sigs.k8s.io/karpenter/pkg/apis/v1beta1"
 	"sigs.k8s.io/karpenter/pkg/scheduling"
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 	"sigs.k8s.io/karpenter/pkg/utils/pretty"
 
 	"sigs.k8s.io/karpenter/pkg/cloudprovider"
@@ -66,16 +66,12 @@ type LaunchOptions struct {
 }
 
 // RecordPodNomination causes nominate pod events to be recorded against the node.
-func RecordPodNomination(o LaunchOptions) LaunchOptions {
+func RecordPodNomination(o *LaunchOptions) {
 	o.RecordPodNomination = true
-	return o
 }
 
-func WithReason(reason string) func(LaunchOptions) LaunchOptions {
-	return func(o LaunchOptions) LaunchOptions {
-		o.Reason = reason
-		return o
-	}
+func WithReason(reason string) func(*LaunchOptions) {
+	return func(o *LaunchOptions) { o.Reason = reason }
 }
 
 // Provisioner waits for enqueued pods, batches them, creates capacity and binds the pods to the capacity.
@@ -146,7 +142,7 @@ func (p *Provisioner) Reconcile(ctx context.Context) (result reconcile.Result, e
 
 // CreateNodeClaims launches nodes passed into the function in parallel. It returns a slice of the successfully created node
 // names as well as a multierr of any errors that occurred while launching nodes
-func (p *Provisioner) CreateNodeClaims(ctx context.Context, nodeClaims []*scheduler.NodeClaim, opts ...functional.Option[LaunchOptions]) ([]string, error) {
+func (p *Provisioner) CreateNodeClaims(ctx context.Context, nodeClaims []*scheduler.NodeClaim, opts ...option.Function[LaunchOptions]) ([]string, error) {
 	// Create capacity and bind pods
 	errs := make([]error, len(nodeClaims))
 	nodeClaimNames := make([]string, len(nodeClaims))
@@ -362,9 +358,9 @@ func (p *Provisioner) Schedule(ctx context.Context) (scheduler.Results, error) {
 	return results, nil
 }
 
-func (p *Provisioner) Create(ctx context.Context, n *scheduler.NodeClaim, opts ...functional.Option[LaunchOptions]) (string, error) {
+func (p *Provisioner) Create(ctx context.Context, n *scheduler.NodeClaim, opts ...option.Function[LaunchOptions]) (string, error) {
 	ctx = log.IntoContext(ctx, log.FromContext(ctx).WithValues("NodePool", klog.KRef("", n.NodePoolName)))
-	options := functional.ResolveOptions(opts...)
+	options := option.Resolve(opts...)
 	latest := &v1beta1.NodePool{}
 	if err := p.kubeClient.Get(ctx, types.NamespacedName{Name: n.NodePoolName}, latest); err != nil {
 		return "", fmt.Errorf("getting current resource usage, %w", err)
@@ -394,7 +390,7 @@ func (p *Provisioner) Create(ctx context.Context, n *scheduler.NodeClaim, opts .
 	// to then trigger cluster state updates. Triggering it manually ensures that Karpenter waits for the
 	// internal cache to sync before moving onto another disruption loop.
 	p.cluster.UpdateNodeClaim(nodeClaim)
-	if functional.ResolveOptions(opts...).RecordPodNomination {
+	if option.Resolve(opts...).RecordPodNomination {
 		for _, pod := range n.Pods {
 			p.recorder.Publish(scheduler.NominatePodEvent(pod, nil, nodeClaim))
 		}

pkg/controllers/provisioning/scheduling/topology.go

@@ -21,11 +21,11 @@ import (
 	"fmt"
 	"math"
 
+	"github.com/awslabs/operatorpkg/option"
 	"k8s.io/apimachinery/pkg/api/errors"
 
 	"sigs.k8s.io/karpenter/pkg/controllers/state"
 	"sigs.k8s.io/karpenter/pkg/scheduling"
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 	"sigs.k8s.io/karpenter/pkg/utils/pretty"
 
 	"go.uber.org/multierr"
@@ -136,7 +136,7 @@ func (t *Topology) Update(ctx context.Context, p *v1.Pod) error {
 }
 
 // Record records the topology changes given that pod p schedule on a node with the given requirements
-func (t *Topology) Record(p *v1.Pod, requirements scheduling.Requirements, compatabilityOptions ...functional.Option[scheduling.CompatibilityOptions]) {
+func (t *Topology) Record(p *v1.Pod, requirements scheduling.Requirements, compatabilityOptions ...option.Function[scheduling.CompatibilityOptions]) {
 	// once we've committed to a domain, we record the usage in every topology that cares about it
 	for _, tc := range t.topologies {
 		if tc.Counts(p, requirements, compatabilityOptions...) {
@@ -165,7 +165,7 @@ func (t *Topology) Record(p *v1.Pod, requirements scheduling.Requirements, compa
 // affinities, anti-affinities or inverse anti-affinities.  The nodeHostname is the hostname that we are currently considering
 // placing the pod on.  It returns these newly tightened requirements, or an error in the case of a set of requirements that
 // cannot be satisfied.
-func (t *Topology) AddRequirements(podRequirements, nodeRequirements scheduling.Requirements, p *v1.Pod, compatabilityOptions ...functional.Option[scheduling.CompatibilityOptions]) (scheduling.Requirements, error) {
+func (t *Topology) AddRequirements(podRequirements, nodeRequirements scheduling.Requirements, p *v1.Pod, compatabilityOptions ...option.Function[scheduling.CompatibilityOptions]) (scheduling.Requirements, error) {
 	requirements := scheduling.NewRequirements(nodeRequirements.Values()...)
 	for _, topology := range t.getMatchingTopologies(p, nodeRequirements, compatabilityOptions...) {
 		podDomains := scheduling.NewRequirement(topology.Key, v1.NodeSelectorOpExists)
@@ -381,7 +381,7 @@ func (t *Topology) buildNamespaceList(ctx context.Context, namespace string, nam
 
 // getMatchingTopologies returns a sorted list of topologies that either control the scheduling of pod p, or for which
 // the topology selects pod p and the scheduling of p affects the count per topology domain
-func (t *Topology) getMatchingTopologies(p *v1.Pod, requirements scheduling.Requirements, compatabilityOptions ...functional.Option[scheduling.CompatibilityOptions]) []*TopologyGroup {
+func (t *Topology) getMatchingTopologies(p *v1.Pod, requirements scheduling.Requirements, compatabilityOptions ...option.Function[scheduling.CompatibilityOptions]) []*TopologyGroup {
 	var matchingTopologies []*TopologyGroup
 	for _, tc := range t.topologies {
 		if tc.IsOwnedBy(p.UID) {

pkg/controllers/provisioning/scheduling/topologygroup.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"math"
 
+	"github.com/awslabs/operatorpkg/option"
 	"github.com/mitchellh/hashstructure/v2"
 	"github.com/samber/lo"
 	v1 "k8s.io/api/core/v1"
@@ -29,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	"sigs.k8s.io/karpenter/pkg/scheduling"
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 type TopologyType byte
@@ -114,7 +114,7 @@ func (t *TopologyGroup) Record(domains ...string) {
 
 // Counts returns true if the pod would count for the topology, given that it schedule to a node with the provided
 // requirements
-func (t *TopologyGroup) Counts(pod *v1.Pod, requirements scheduling.Requirements, compatabilityOptions ...functional.Option[scheduling.CompatibilityOptions]) bool {
+func (t *TopologyGroup) Counts(pod *v1.Pod, requirements scheduling.Requirements, compatabilityOptions ...option.Function[scheduling.CompatibilityOptions]) bool {
 	return t.selects(pod) && t.nodeFilter.MatchesRequirements(requirements, compatabilityOptions...)
 }
 

pkg/controllers/provisioning/scheduling/topologynodefilter.go

@@ -17,10 +17,10 @@ limitations under the License.
 package scheduling
 
 import (
+	"github.com/awslabs/operatorpkg/option"
 	v1 "k8s.io/api/core/v1"
 
 	"sigs.k8s.io/karpenter/pkg/scheduling"
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 // TopologyNodeFilter is used to determine if a given actual node or scheduling node matches the pod's node selectors
@@ -58,7 +58,7 @@ func (t TopologyNodeFilter) Matches(node *v1.Node) bool {
 // MatchesRequirements returns true if the TopologyNodeFilter doesn't prohibit a node with the requirements from
 // participating in the topology. This method allows checking the requirements from a scheduling.NodeClaim to see if the
 // node we will soon create participates in this topology.
-func (t TopologyNodeFilter) MatchesRequirements(requirements scheduling.Requirements, compatabilityOptions ...functional.Option[scheduling.CompatibilityOptions]) bool {
+func (t TopologyNodeFilter) MatchesRequirements(requirements scheduling.Requirements, compatabilityOptions ...option.Function[scheduling.CompatibilityOptions]) bool {
 	// no requirements, so it always matches
 	if len(t) == 0 {
 		return true

pkg/scheduling/requirements.go

@@ -22,13 +22,13 @@ import (
 	"sort"
 	"strings"
 
+	"github.com/awslabs/operatorpkg/option"
 	"github.com/samber/lo"
 	"go.uber.org/multierr"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 
 	"sigs.k8s.io/karpenter/pkg/apis/v1beta1"
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 // Requirements are an efficient set representation under the hood. Since its underlying
@@ -163,18 +163,18 @@ type CompatibilityOptions struct {
 	AllowUndefined sets.Set[string]
 }
 
-var AllowUndefinedWellKnownLabels = func(options CompatibilityOptions) CompatibilityOptions {
+var AllowUndefinedWellKnownLabels = func(options *CompatibilityOptions) {
 	options.AllowUndefined = v1beta1.WellKnownLabels
-	return options
 }
 
-func (r Requirements) IsCompatible(requirements Requirements, options ...functional.Option[CompatibilityOptions]) bool {
+func (r Requirements) IsCompatible(requirements Requirements, options ...option.Function[CompatibilityOptions]) bool {
 	return r.Compatible(requirements, options...) == nil
 }
 
 // Compatible ensures the provided requirements can loosely be met.
-func (r Requirements) Compatible(requirements Requirements, options ...functional.Option[CompatibilityOptions]) (errs error) {
-	opts := functional.ResolveOptions(options...)
+func (r Requirements) Compatible(requirements Requirements, options ...option.Function[CompatibilityOptions]) (errs error) {
+	opts := option.Resolve(options...)
+
 	// Custom Labels must intersect, but if not defined are denied.
 	for key := range requirements.Keys().Difference(opts.AllowUndefined) {
 		if operator := requirements.Get(key).Operator(); r.Has(key) || operator == v1.NodeSelectorOpNotIn || operator == v1.NodeSelectorOpDoesNotExist {

pkg/test/environment.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"strings"
 
+	"github.com/awslabs/operatorpkg/option"
 	"github.com/samber/lo"
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -35,7 +36,6 @@ import (
 
 	"sigs.k8s.io/karpenter/pkg/apis/v1beta1"
 	"sigs.k8s.io/karpenter/pkg/utils/env"
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 type Environment struct {
@@ -54,18 +54,16 @@ type EnvironmentOptions struct {
 }
 
 // WithCRDs registers the specified CRDs to the apiserver for use in testing
-func WithCRDs(crds ...*v1.CustomResourceDefinition) functional.Option[EnvironmentOptions] {
-	return func(o EnvironmentOptions) EnvironmentOptions {
+func WithCRDs(crds ...*v1.CustomResourceDefinition) option.Function[EnvironmentOptions] {
+	return func(o *EnvironmentOptions) {
 		o.crds = append(o.crds, crds...)
-		return o
 	}
 }
 
 // WithFieldIndexers expects a function that indexes fields against the cache such as cache.IndexField(...)
-func WithFieldIndexers(fieldIndexers ...func(cache.Cache) error) functional.Option[EnvironmentOptions] {
-	return func(o EnvironmentOptions) EnvironmentOptions {
+func WithFieldIndexers(fieldIndexers ...func(cache.Cache) error) option.Function[EnvironmentOptions] {
+	return func(o *EnvironmentOptions) {
 		o.fieldIndexers = append(o.fieldIndexers, fieldIndexers...)
-		return o
 	}
 }
 
@@ -77,8 +75,8 @@ func NodeClaimFieldIndexer(ctx context.Context) func(cache.Cache) error {
 	}
 }
 
-func NewEnvironment(options ...functional.Option[EnvironmentOptions]) *Environment {
-	opts := functional.ResolveOptions(options...)
+func NewEnvironment(options ...option.Function[EnvironmentOptions]) *Environment {
+	opts := option.Resolve(options...)
 	ctx, cancel := context.WithCancel(context.Background())
 
 	os.Setenv(system.NamespaceEnvKey, "default")

pkg/test/v1alpha1/testnodeclass.go

@@ -19,19 +19,17 @@ package v1alpha1
 import (
 	_ "embed"
 
-	"github.com/samber/lo"
+	"github.com/awslabs/operatorpkg/object"
 	v1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 //go:generate controller-gen crd object:headerFile="../../../hack/boilerplate.go.txt" paths="./..." output:crd:artifacts:config=crds
 var (
 	//go:embed crds/karpenter.test.sh_testnodeclasses.yaml
 	TestNodeClassCRD []byte
 	CRDs             = []*v1.CustomResourceDefinition{
-		lo.Must(functional.Unmarshal[v1.CustomResourceDefinition](TestNodeClassCRD)),
+		object.Unmarshal[v1.CustomResourceDefinition](TestNodeClassCRD),
 	}
 )
 

pkg/utils/atomic/lazy.go

@@ -20,18 +20,16 @@ import (
 	"context"
 	"sync"
 
+	"github.com/awslabs/operatorpkg/option"
 	"github.com/samber/lo"
-
-	"sigs.k8s.io/karpenter/pkg/utils/functional"
 )
 
 type Options struct {
 	ignoreCache bool
 }
 
-func IgnoreCacheOption(o Options) Options {
+func IgnoreCacheOption(o *Options) {
 	o.ignoreCache = true
-	return o
 }
 
 // Lazy persistently stores a value in memory by evaluating
@@ -51,8 +49,8 @@ func (c *Lazy[T]) Set(v T) {
 
 // TryGet attempts to get a non-nil value from the internal value. If the internal value is nil, the Resolve function
 // will attempt to resolve the value, setting the value to be persistently stored if the resolve of Resolve is non-nil.
-func (c *Lazy[T]) TryGet(ctx context.Context, opts ...functional.Option[Options]) (T, error) {
-	o := functional.ResolveOptions(opts...)
+func (c *Lazy[T]) TryGet(ctx context.Context, opts ...option.Function[Options]) (T, error) {
+	o := option.Resolve(opts...)
 	c.mu.RLock()
 	if c.value != nil && !o.ignoreCache {
 		ret := *c.value