diff --git a/controllers/container_controller.go b/controllers/container_controller.go index 9d82913e..fa3824a7 100644 --- a/controllers/container_controller.go +++ b/controllers/container_controller.go @@ -2336,6 +2336,9 @@ func (r *ContainerReconciler) createEndpoints( log.Error(err, "Could not determine host address and port for container port") return nil, err } + if reserveErr := reserveServiceProducerEndpointPort(ctx, r, serviceProducer, hostAddress, hostPort, log); reserveErr != nil { + return nil, reserveErr + } endpointExists := slices.Any(existingEndpoints, func(ep *apiv1.Endpoint) bool { return ep.Spec.ServiceName == serviceProducer.ServiceName && diff --git a/controllers/endpoint_common.go b/controllers/endpoint_common.go index 89463a05..8137b058 100644 --- a/controllers/endpoint_common.go +++ b/controllers/endpoint_common.go @@ -7,6 +7,8 @@ package controllers import ( "context" + "fmt" + "net/netip" "github.com/go-logr/logr" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -16,8 +18,10 @@ import ( ctrl_client "sigs.k8s.io/controller-runtime/pkg/client" apiv1 "github.com/microsoft/dcp/api/v1" + "github.com/microsoft/dcp/internal/networking" "github.com/microsoft/dcp/pkg/commonapi" "github.com/microsoft/dcp/pkg/logger" + "github.com/microsoft/dcp/pkg/ports" "github.com/microsoft/dcp/pkg/slices" "github.com/microsoft/dcp/pkg/syncmap" ) @@ -69,6 +73,76 @@ func SetupEndpointIndexWithManager(mgr ctrl.Manager) error { }) } +func reserveServiceProducerEndpointPort( + ctx context.Context, + client ctrl_client.Client, + serviceProducer commonapi.ServiceProducer, + address string, + port int32, + log logr.Logger, +) error { + var svc apiv1.Service + if getErr := client.Get(ctx, serviceProducer.ServiceNamespacedName(), &svc); getErr != nil { + if apierrors.IsNotFound(getErr) { + return nil + } + return getErr + } + + ip, ipErr := portReservationIP(address) + if ipErr != nil { + return ipErr + } + return networking.ReserveSpecificPort(ctx, ports.Binding{Protocol: string(svc.Spec.Protocol), IP: ip, Port: port}, log) +} + +func portReservationIP(address string) (netip.Addr, error) { + normalizedAddress, addressErr := networking.NormalizePortAllocationAddress(address) + if addressErr != nil { + return netip.Addr{}, addressErr + } + ip, ipErr := netip.ParseAddr(networking.ToStandaloneAddress(normalizedAddress)) + if ipErr != nil { + return netip.Addr{}, fmt.Errorf("could not parse port reservation IP '%s': %w", normalizedAddress, ipErr) + } + return ip, nil +} + +func releaseEndpointPortReservation(ctx context.Context, client ctrl_client.Client, endpoint *apiv1.Endpoint, log logr.Logger) error { + var svc apiv1.Service + if getErr := client.Get(ctx, types.NamespacedName{Namespace: endpoint.Spec.ServiceNamespace, Name: endpoint.Spec.ServiceName}, &svc); getErr != nil { + if !apierrors.IsNotFound(getErr) { + log.Error(getErr, "Could not get Service for Endpoint port release", + "Endpoint", endpoint.NamespacedName().String(), + "ServiceNamespace", endpoint.Spec.ServiceNamespace, + "ServiceName", endpoint.Spec.ServiceName, + ) + return getErr + } + return nil + } + + return releaseEndpointPortReservationForProtocol(ctx, svc.Spec.Protocol, endpoint, log) +} + +func releaseEndpointPortReservationForProtocol(ctx context.Context, protocol apiv1.PortProtocol, endpoint *apiv1.Endpoint, log logr.Logger) error { + ip, ipErr := portReservationIP(endpoint.Spec.Address) + if ipErr != nil { + return ipErr + } + + releaseErr := networking.ReleaseSpecificPort(ctx, ports.Binding{Protocol: string(protocol), IP: ip, Port: endpoint.Spec.Port}, log) + if releaseErr != nil { + log.Error(releaseErr, "Could not release Endpoint port reservation", + "Endpoint", endpoint.NamespacedName().String(), + "Address", endpoint.Spec.Address, + "Port", endpoint.Spec.Port, + ) + return releaseErr + } + return nil +} + // Attempts to create Endpoint objects for all Services exposed by the given workload object (owner parameter). // If reservedServicePorts is not nil, it is used to determine the port to use for each Service. // The ecc parameter contains additional context data needed for Endpoint creation, @@ -148,6 +222,8 @@ func ensureEndpointsForWorkload[EndpointCreationContext any]( "Endpoint", endpoint.NamespacedName().String(), ) existingEndpoints = append(existingEndpoints, endpoint) + } else { + _ = releaseEndpointPortReservation(ctx, r, endpoint, svcLog) } } } @@ -174,6 +250,7 @@ func ensureEndpointsForWorkload[EndpointCreationContext any]( for _, endpoint := range newEndpoints { if err = ctrl.SetControllerReference(owner, endpoint, r.Scheme()); err != nil { svcLog.Error(err, "Failed to set owner for Endpoint") + _ = releaseEndpointPortReservation(ctx, r, endpoint, svcLog) continue // Try to persist other endpoints } @@ -182,6 +259,7 @@ func ensureEndpointsForWorkload[EndpointCreationContext any]( if !apiv1.ResourceCreationProhibited.Load() { svcLog.Error(err, "Could not persist Endpoint object") } + _ = releaseEndpointPortReservation(ctx, r, endpoint, svcLog) continue } @@ -214,6 +292,8 @@ func removeEndpointsForWorkload(ctx context.Context, r ctrl_client.Client, owner "Endpoint", endpoint.NamespacedName().String(), "Workload", owner.NamespacedName().String(), ) + } else { + _ = releaseEndpointPortReservation(ctx, r, &endpoint, log) } sweKey := ServiceWorkloadEndpointKey{ diff --git a/controllers/executable_controller.go b/controllers/executable_controller.go index b0e16c0b..0bb5bcc4 100644 --- a/controllers/executable_controller.go +++ b/controllers/executable_controller.go @@ -1019,6 +1019,10 @@ func (r *ExecutableReconciler) createEndpoints( address = networking.IPv6LocalhostDefaultAddress } + if reserveErr := reserveServiceProducerEndpointPort(ctx, r, serviceProducer, address, serviceProducer.Port, log); reserveErr != nil { + return nil, reserveErr + } + endpointExists := slices.Any(existingEndpoints, func(ep *apiv1.Endpoint) bool { return ep.Spec.ServiceName == serviceProducer.ServiceName && ep.Spec.ServiceNamespace == owner.GetNamespace() && diff --git a/controllers/service_controller.go b/controllers/service_controller.go index db33054f..de2cacde 100644 --- a/controllers/service_controller.go +++ b/controllers/service_controller.go @@ -11,6 +11,7 @@ import ( "fmt" "math/rand" "net" + "net/netip" "os" "path/filepath" "sync/atomic" @@ -36,14 +37,16 @@ import ( usvc_io "github.com/microsoft/dcp/pkg/io" "github.com/microsoft/dcp/pkg/logger" "github.com/microsoft/dcp/pkg/pointers" + "github.com/microsoft/dcp/pkg/ports" "github.com/microsoft/dcp/pkg/process" "github.com/microsoft/dcp/pkg/slices" "github.com/microsoft/dcp/pkg/syncmap" ) type proxyInstanceData struct { - proxy proxy.Proxy - stopProxy context.CancelFunc + proxy proxy.Proxy + stopProxy context.CancelFunc + portReservation ports.Binding } type serviceData struct { @@ -166,7 +169,7 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct if err := r.Get(ctx, req.NamespacedName, &svc); err != nil { if apimachinery_errors.IsNotFound(err) { log.V(1).Info("The Service object does not exist yet or was deleted") - r.stopService(req.NamespacedName, nil, log) + _ = r.stopService(ctx, req.NamespacedName, nil, log) getNotFoundCounter.Add(ctx, 1) return ctrl.Result{}, nil } else { @@ -185,11 +188,16 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct if svc.DeletionTimestamp != nil && !svc.DeletionTimestamp.IsZero() { log.V(1).Info("Service object is being deleted") - r.stopService(svc.NamespacedName(), &svc, log) - change = deleteFinalizer(&svc, serviceFinalizer, log) - serviceCounters(ctx, &svc, -1) // Service is being deleted + change = r.stopService(ctx, svc.NamespacedName(), &svc, log) + change |= releaseDeclaredServicePort(ctx, &svc, log) + change |= r.releaseServiceEndpointPortReservations(ctx, &svc, log) + if change == noChange { + change = deleteFinalizer(&svc, serviceFinalizer, log) + serviceCounters(ctx, &svc, -1) // Service is being deleted + } } else { change = ensureFinalizer(&svc, serviceFinalizer, log) + change |= reserveDeclaredServicePort(ctx, &svc, log) // If we added a finalizer, we'll do the additional reconciliation next call if change == noChange { change |= r.ensureServiceEffectiveAddressAndPort(ctx, &svc, log) @@ -202,15 +210,79 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct return result, err } -func (r *ServiceReconciler) stopService(svcName types.NamespacedName, svc *apiv1.Service, log logr.Logger) { - serviceData, found := r.serviceInfo.LoadAndDelete(svcName) +func reserveDeclaredServicePort(ctx context.Context, svc *apiv1.Service, log logr.Logger) objectChange { + if svc.Spec.Port == 0 { + return noChange + } + + requestedServiceAddress, requestedAddressErr := getRequestedServiceAddress(svc) + if requestedAddressErr != nil { + log.Error(requestedAddressErr, "Could not determine requested service address for port reservation") + return additionalReconciliationNeeded + } + + requestedServiceIP, requestedServiceIPErr := portReservationIP(requestedServiceAddress) + if requestedServiceIPErr != nil { + log.Error(requestedServiceIPErr, "Could not determine requested service IP for port reservation") + return additionalReconciliationNeeded + } + + reserveErr := networking.ReserveSpecificPort(ctx, ports.Binding{Protocol: string(svc.Spec.Protocol), IP: requestedServiceIP, Port: svc.Spec.Port}, log) + if reserveErr != nil { + log.Error(reserveErr, "Could not reserve declared service port", + "Address", requestedServiceAddress, + "Port", svc.Spec.Port, + ) + return additionalReconciliationNeeded + } + + return noChange +} + +func releaseDeclaredServicePort(ctx context.Context, svc *apiv1.Service, log logr.Logger) objectChange { + if svc.Spec.Port == 0 { + return noChange + } + + requestedServiceAddress, requestedAddressErr := getRequestedServiceAddress(svc) + if requestedAddressErr != nil { + log.Error(requestedAddressErr, "Could not determine requested service address for port release") + return additionalReconciliationNeeded + } + + requestedServiceIP, requestedServiceIPErr := portReservationIP(requestedServiceAddress) + if requestedServiceIPErr != nil { + log.Error(requestedServiceIPErr, "Could not determine requested service IP for port release") + return additionalReconciliationNeeded + } + + releaseErr := networking.ReleaseSpecificPort(ctx, ports.Binding{Protocol: string(svc.Spec.Protocol), IP: requestedServiceIP, Port: svc.Spec.Port}, log) + if releaseErr != nil { + log.Error(releaseErr, "Could not release declared service port", + "Address", requestedServiceAddress, + "Port", svc.Spec.Port, + ) + return additionalReconciliationNeeded + } + + return noChange +} + +func (r *ServiceReconciler) stopService(ctx context.Context, svcName types.NamespacedName, svc *apiv1.Service, log logr.Logger) objectChange { + serviceData, found := r.serviceInfo.Load(svcName) if !found || len(serviceData.proxies) == 0 { - return + return noChange + } + releaseErr := stopProxiesAndReleasePortReservations(ctx, serviceData.proxies, log) + if releaseErr != nil { + log.Error(releaseErr, "Could not release service proxy port reservation(s)") + return additionalReconciliationNeeded } - stopProxies(serviceData.proxies, log) + r.serviceInfo.Delete(svcName) if svc != nil { logger.ReleaseResourceLog(svc.GetResourceId()) } + return noChange } func (r *ServiceReconciler) ensureServiceEffectiveAddressAndPort(ctx context.Context, svc *apiv1.Service, log logr.Logger) objectChange { @@ -379,11 +451,31 @@ func (r *ServiceReconciler) getServiceEndpoints(ctx context.Context, svc *apiv1. } } +func (r *ServiceReconciler) releaseServiceEndpointPortReservations(ctx context.Context, svc *apiv1.Service, log logr.Logger) objectChange { + serviceEndpoints, serviceEndpointsErr := r.getServiceEndpoints(ctx, svc, log) + if serviceEndpointsErr != nil { + return additionalReconciliationNeeded + } + + var releaseErr error + for endpointIndex := range serviceEndpoints.Items { + endpoint := &serviceEndpoints.Items[endpointIndex] + endpointReleaseErr := releaseEndpointPortReservationForProtocol(ctx, svc.Spec.Protocol, endpoint, log) + if endpointReleaseErr != nil { + releaseErr = errors.Join(releaseErr, endpointReleaseErr) + } + } + if releaseErr != nil { + return additionalReconciliationNeeded + } + return noChange +} + // startProxyIfNeeded() starts all necessary proxies for the given service. // No matter if an error occurs, a *serviceData is returned that is used for storing proxy information // and count startup attempts for the service. The service data is stored in ServiceReconciler.serviceInfo // before the method returns. -func (r *ServiceReconciler) startProxyIfNeeded(_ context.Context, svc *apiv1.Service, log logr.Logger) (*serviceData, error) { +func (r *ServiceReconciler) startProxyIfNeeded(ctx context.Context, svc *apiv1.Service, log logr.Logger) (*serviceData, error) { psd, found := r.serviceInfo.Load(svc.NamespacedName()) if !found { psd = &serviceData{} @@ -400,26 +492,33 @@ func (r *ServiceReconciler) startProxyIfNeeded(_ context.Context, svc *apiv1.Ser svc.Status.ProxyProcessPid = apiv1.UnknownPID svc.Status.EffectiveAddress = "" svc.Status.EffectivePort = 0 - stopProxies(psd.proxies, log) + defer r.serviceInfo.Store(svc.NamespacedName(), psd) + + releaseExistingErr := stopProxiesAndReleasePortReservations(ctx, psd.proxies, log) + if releaseExistingErr != nil { + return psd, fmt.Errorf("could not release existing service proxy port reservation(s): %w", releaseExistingErr) + } psd.proxies = nil psd.startAttempts += 1 - defer r.serviceInfo.Store(svc.NamespacedName(), psd) if requestedAddressErr != nil { return psd, requestedAddressErr } - proxies, portAllocationErr := r.getProxyData(svc, requestedServiceAddress, log) + proxies, portAllocationErr := r.getProxyData(ctx, svc, requestedServiceAddress, log) if portAllocationErr != nil { - stopProxies(proxies, log) + releaseErr := stopProxiesAndReleasePortReservations(ctx, proxies, log) + if releaseErr != nil { + portAllocationErr = errors.Join(portAllocationErr, releaseErr) + } return psd, fmt.Errorf("could not create the proxy for the service: port allocation failed: %w", portAllocationErr) } for _, proxyInstanceData := range proxies { startErr := proxyInstanceData.proxy.Start() if startErr != nil { - stopProxies(proxies, log) - return psd, fmt.Errorf("could not start the proxy for the service: %w", startErr) + releaseErr := stopProxiesAndReleasePortReservations(ctx, proxies, log) + return psd, fmt.Errorf("could not start the proxy for the service: %w", errors.Join(startErr, releaseErr)) } else { log.V(1).Info("Service proxy started", "ProxyListenAddress", proxyInstanceData.proxy.ListenAddress(), @@ -442,12 +541,12 @@ func (r *ServiceReconciler) startProxyIfNeeded(_ context.Context, svc *apiv1.Ser } // Allocates (but does not start) proxy instances for the given service. -func (r *ServiceReconciler) getProxyData(svc *apiv1.Service, requestedServiceAddress string, log logr.Logger) ([]proxyInstanceData, error) { +func (r *ServiceReconciler) getProxyData(ctx context.Context, svc *apiv1.Service, requestedServiceAddress string, log logr.Logger) ([]proxyInstanceData, error) { proxies := []proxyInstanceData{} var getProxyPort func(proxyAddress string) (int32, error) if svc.Spec.Port == 0 { getProxyPort = func(proxyAddress string) (int32, error) { - return networking.GetFreePort(svc.Spec.Protocol, proxyAddress, log) + return networking.GetFreePort(ctx, svc.Spec.Protocol, proxyAddress, log) } } else { getProxyPort = func(_ string) (int32, error) { @@ -490,7 +589,7 @@ func (r *ServiceReconciler) getProxyData(svc *apiv1.Service, requestedServiceAdd if usingSamePort { proxyPort = lastPort - err = networking.CheckPortAvailable(svc.Spec.Protocol, proxyInstanceAddress, proxyPort, log) + err = networking.CheckPortAvailable(ctx, svc.Spec.Protocol, proxyInstanceAddress, proxyPort, log) if err != nil { usingSamePort = false log.Info("Could not use the same port for all addresses associated with requested service address, service will be reachable only using specific IP address", @@ -508,10 +607,21 @@ func (r *ServiceReconciler) getProxyData(svc *apiv1.Service, requestedServiceAdd } } + proxyReservationIP, proxyReservationIPErr := netip.ParseAddr(networking.ToStandaloneAddress(proxyInstanceAddress)) + if proxyReservationIPErr != nil { + portAllocationErr = errors.Join(portAllocationErr, fmt.Errorf("could not parse proxy reservation IP '%s': %w", proxyInstanceAddress, proxyReservationIPErr)) + continue + } + proxyCtx, cancelFunc := context.WithCancel(r.LifetimeCtx) proxies = append(proxies, proxyInstanceData{ proxy: r.config.CreateProxy(svc.Spec.Protocol, proxyInstanceAddress, proxyPort, proxyCtx, proxyLog), stopProxy: cancelFunc, + portReservation: ports.Binding{ + Protocol: string(svc.Spec.Protocol), + IP: proxyReservationIP, + Port: proxyPort, + }, }) } @@ -535,7 +645,7 @@ func (r *ServiceReconciler) getEffectiveAddressAndPort(proxies []proxyInstanceDa // For any "hostname" style addresses (e.g. "localhost", ""), we use "localhost" as the effective // address if possible. - if net.ParseIP(networking.ToStandaloneAddress(requestedServiceAddress)) == nil { + if _, parseErr := netip.ParseAddr(networking.ToStandaloneAddress(requestedServiceAddress)); parseErr != nil { portsInUse := make(map[int32]bool) for _, pd := range eligibleProxies { portsInUse[pd.proxy.EffectivePort()] = true @@ -624,3 +734,27 @@ func stopProxies(proxies []proxyInstanceData, log logr.Logger) { } } } + +func stopProxiesAndReleasePortReservations(ctx context.Context, proxies []proxyInstanceData, log logr.Logger) error { + var releaseErr error + for _, proxyInstanceData := range proxies { + proxyReleaseErr := networking.ReleaseSpecificPort( + ctx, + proxyInstanceData.portReservation, + log, + ) + if proxyReleaseErr != nil { + log.Error(proxyReleaseErr, "Could not release service proxy port reservation", + "IP", proxyInstanceData.portReservation.IP, + "Port", proxyInstanceData.portReservation.Port, + ) + releaseErr = errors.Join(releaseErr, proxyReleaseErr) + } + } + if releaseErr != nil { + return releaseErr + } + + stopProxies(proxies, log) + return nil +} diff --git a/controllers/service_controller_test.go b/controllers/service_controller_test.go new file mode 100644 index 00000000..121a3b9c --- /dev/null +++ b/controllers/service_controller_test.go @@ -0,0 +1,344 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package controllers + +import ( + "context" + "errors" + "net/netip" + "path/filepath" + "testing" + "time" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + ctrl_client "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + apiv1 "github.com/microsoft/dcp/api/v1" + "github.com/microsoft/dcp/internal/dcpclient" + "github.com/microsoft/dcp/internal/networking" + "github.com/microsoft/dcp/internal/statestore" + internal_testutil "github.com/microsoft/dcp/internal/testutil" + "github.com/microsoft/dcp/pkg/commonapi" + usvc_io "github.com/microsoft/dcp/pkg/io" + "github.com/microsoft/dcp/pkg/osutil" + "github.com/microsoft/dcp/pkg/ports" + "github.com/microsoft/dcp/pkg/process" + "github.com/microsoft/dcp/pkg/syncmap" + "github.com/microsoft/dcp/pkg/testutil" +) + +func TestStopServiceReleasesProxyPortReservations(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, 10*time.Second) + defer cancel() + log := logr.Discard() + store, owner := configureTestPortAllocator(t, ctx) + + const address = networking.IPv4LocalhostDefaultAddress + const port int32 = 26030 + reserveTestPort(t, ctx, store, owner, apiv1.TCP, address, port) + requirePortReserved(t, ctx, store, apiv1.TCP, address, port, owner) + + serviceName := types.NamespacedName{Name: "test-service", Namespace: metav1.NamespaceNone} + stopped := false + r := &ServiceReconciler{ + serviceInfo: &syncmap.Map[types.NamespacedName, *serviceData]{}, + } + r.serviceInfo.Store(serviceName, &serviceData{ + proxies: []proxyInstanceData{ + { + stopProxy: func() { stopped = true }, + portReservation: ports.Binding{ + Protocol: string(apiv1.TCP), + IP: netip.MustParseAddr(address), + Port: port, + }, + }, + }, + }) + + change := r.stopService(ctx, serviceName, nil, log) + + require.Equal(t, noChange, change) + require.True(t, stopped) + _, found := r.serviceInfo.Load(serviceName) + require.False(t, found) + requirePortReleased(t, ctx, store, apiv1.TCP, address, port, owner) +} + +func TestStopServiceDoesNotStopProxyWhenReservationReleaseFails(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, 10*time.Second) + defer cancel() + log := logr.Discard() + configureTestPortAllocator(t, ctx) + + serviceName := types.NamespacedName{Name: "test-service-release-failure", Namespace: metav1.NamespaceNone} + stopped := false + r := &ServiceReconciler{ + serviceInfo: &syncmap.Map[types.NamespacedName, *serviceData]{}, + } + r.serviceInfo.Store(serviceName, &serviceData{ + proxies: []proxyInstanceData{ + { + stopProxy: func() { stopped = true }, + portReservation: ports.Binding{ + Protocol: string(apiv1.TCP), + IP: netip.MustParseAddr(networking.IPv4LocalhostDefaultAddress), + Port: networking.InvalidPort, + }, + }, + }, + }) + + change := r.stopService(ctx, serviceName, nil, log) + + require.Equal(t, additionalReconciliationNeeded, change) + require.False(t, stopped) + updatedServiceData, found := r.serviceInfo.Load(serviceName) + require.True(t, found) + require.Len(t, updatedServiceData.proxies, 1) +} + +func TestServiceDeletionReleasesEndpointPortReservations(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, 10*time.Second) + defer cancel() + log := logr.Discard() + store, owner := configureTestPortAllocator(t, ctx) + + const address = networking.IPv4LocalhostDefaultAddress + const port int32 = 26031 + svc := &apiv1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-proxyless-service", + Namespace: metav1.NamespaceNone, + }, + Spec: apiv1.ServiceSpec{ + Protocol: apiv1.TCP, + AddressAllocationMode: apiv1.AddressAllocationModeProxyless, + }, + } + endpoint := &apiv1.Endpoint{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-proxyless-endpoint", + Namespace: metav1.NamespaceNone, + }, + Spec: apiv1.EndpointSpec{ + ServiceNamespace: svc.Namespace, + ServiceName: svc.Name, + Address: address, + Port: port, + }, + } + reserveTestPort(t, ctx, store, owner, svc.Spec.Protocol, endpoint.Spec.Address, endpoint.Spec.Port) + requirePortReserved(t, ctx, store, svc.Spec.Protocol, endpoint.Spec.Address, endpoint.Spec.Port, owner) + + apiClient := fake.NewClientBuilder(). + WithScheme(dcpclient.NewScheme()). + WithObjects(svc, endpoint). + WithIndex(&apiv1.Endpoint{}, serviceNamespaceKey, func(rawObj ctrl_client.Object) []string { + indexedEndpoint := rawObj.(*apiv1.Endpoint) + return []string{indexedEndpoint.Spec.ServiceNamespace} + }). + WithIndex(&apiv1.Endpoint{}, serviceNameKey, func(rawObj ctrl_client.Object) []string { + indexedEndpoint := rawObj.(*apiv1.Endpoint) + return []string{indexedEndpoint.Spec.ServiceName} + }). + Build() + processExecutor := internal_testutil.NewTestProcessExecutor(ctx) + defer processExecutor.Dispose() + r := NewServiceReconciler(ctx, apiClient, apiClient, log, ServiceReconcilerConfig{ProcessExecutor: processExecutor}) + + change := r.releaseServiceEndpointPortReservations(ctx, svc, log) + + require.Equal(t, noChange, change) + requirePortReleased(t, ctx, store, svc.Spec.Protocol, endpoint.Spec.Address, endpoint.Spec.Port, owner) +} + +func TestEnsureEndpointsForWorkloadReleasesReservationWhenCreateFails(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, 10*time.Second) + defer cancel() + log := logr.Discard() + store, owner := configureTestPortAllocator(t, ctx) + + const serviceName = "test-endpoint-create-failure-service" + const address = networking.IPv4LocalhostDefaultAddress + const port int32 = 26032 + svc := &apiv1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: serviceName, + Namespace: metav1.NamespaceNone, + }, + Spec: apiv1.ServiceSpec{ + Protocol: apiv1.TCP, + }, + } + workload := &apiv1.Container{ + TypeMeta: metav1.TypeMeta{ + APIVersion: apiv1.GroupVersion.String(), + Kind: "Container", + }, + ObjectMeta: metav1.ObjectMeta{ + Name: "test-endpoint-create-failure-container", + Namespace: metav1.NamespaceNone, + UID: "test-endpoint-create-failure-container", + Annotations: map[string]string{ + commonapi.ServiceProducerAnnotation: `[{"serviceName":"` + serviceName + `","port":80}]`, + }, + }, + } + endpoint := &apiv1.Endpoint{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-endpoint-create-failure-endpoint", + Namespace: metav1.NamespaceNone, + }, + Spec: apiv1.EndpointSpec{ + ServiceNamespace: svc.Namespace, + ServiceName: svc.Name, + Address: address, + Port: port, + }, + } + apiClient := fake.NewClientBuilder(). + WithScheme(dcpclient.NewScheme()). + WithObjects(svc). + WithIndex(&apiv1.Endpoint{}, commonapi.WorkloadOwnerKey, func(rawObj ctrl_client.Object) []string { + indexedEndpoint := rawObj.(*apiv1.Endpoint) + owners := indexedEndpoint.GetOwnerReferences() + ownerUIDs := make([]string, 0, len(owners)) + for _, ownerReference := range owners { + ownerUIDs = append(ownerUIDs, string(ownerReference.UID)) + } + return ownerUIDs + }). + Build() + createErr := errors.New("create failed") + endpointOwner := &createFailingEndpointOwner{ + Client: apiClient, + endpointFactory: func(ctx context.Context, _ ctrl_client.Object, _ commonapi.ServiceProducer, _ []*apiv1.Endpoint, _ struct{}, _ logr.Logger) ([]*apiv1.Endpoint, error) { + reserveTestPort(t, ctx, store, owner, svc.Spec.Protocol, endpoint.Spec.Address, endpoint.Spec.Port) + return []*apiv1.Endpoint{endpoint}, nil + }, + createErr: createErr, + } + + ensureEndpointsForWorkload(ctx, endpointOwner, workload, nil, struct{}{}, log) + + requirePortReleased(t, ctx, store, svc.Spec.Protocol, endpoint.Spec.Address, endpoint.Spec.Port, owner) +} + +type createFailingEndpointOwner struct { + ctrl_client.Client + endpointFactory func(context.Context, ctrl_client.Object, commonapi.ServiceProducer, []*apiv1.Endpoint, struct{}, logr.Logger) ([]*apiv1.Endpoint, error) + createErr error +} + +func (o *createFailingEndpointOwner) Create(ctx context.Context, obj ctrl_client.Object, opts ...ctrl_client.CreateOption) error { + return o.createErr +} + +func (o *createFailingEndpointOwner) createEndpoints( + ctx context.Context, + owner ctrl_client.Object, + serviceProducer commonapi.ServiceProducer, + existingEndpoints []*apiv1.Endpoint, + ecc struct{}, + log logr.Logger, +) ([]*apiv1.Endpoint, error) { + return o.endpointFactory(ctx, owner, serviceProducer, existingEndpoints, ecc, log) +} + +func (o *createFailingEndpointOwner) validateExistingEndpoints( + ctx context.Context, + owner ctrl_client.Object, + serviceProducer commonapi.ServiceProducer, + endpoints []*apiv1.Endpoint, + ecc struct{}, + log logr.Logger, +) ([]*apiv1.Endpoint, []*apiv1.Endpoint, error) { + return endpoints, nil, nil +} + +func configureTestPortAllocator(t *testing.T, ctx context.Context) (*statestore.Store, process.ProcessTreeItem) { + t.Helper() + + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + require.NoError(t, usvc_io.EnsureRestrictedDirectory(filepath.Dir(storePath), osutil.PermissionOnlyOwnerReadWriteTraverse)) + store, openErr := statestore.Open(ctx, statestore.Options{Path: storePath}) + require.NoError(t, openErr) + t.Cleanup(func() { + require.NoError(t, store.Close()) + }) + + owner, ownerErr := statestore.CurrentResourceLeaseOwner() + require.NoError(t, ownerErr) + networking.ConfigureStateStorePortAllocator(store, owner) + t.Cleanup(func() { + networking.ConfigureStateStorePortAllocator(nil, process.ProcessTreeItem{}) + }) + t.Setenv(networking.DCP_PORT_ALLOCATOR, "statestore") + return store, owner +} + +func reserveTestPort( + t *testing.T, + ctx context.Context, + store *statestore.Store, + owner process.ProcessTreeItem, + protocol apiv1.PortProtocol, + address string, + port int32, +) { + t.Helper() + + _, reserveErr := store.CreateOrUpdatePortReservation(ctx, statestore.PortReservationRequest{ + Binding: ports.Binding{Protocol: string(protocol), IP: netip.MustParseAddr(networking.ToStandaloneAddress(address)), Port: port}, + OwnerProcess: owner, + }) + require.NoError(t, reserveErr) +} + +func requirePortReserved( + t *testing.T, + ctx context.Context, + store *statestore.Store, + protocol apiv1.PortProtocol, + address string, + port int32, + owner process.ProcessTreeItem, +) { + t.Helper() + + otherOwner := owner + otherOwner.IdentityTime = otherOwner.IdentityTime.Add(time.Second) + _, reserveErr := store.CreatePortReservation(ctx, statestore.PortReservationRequest{ + Binding: ports.Binding{Protocol: string(protocol), IP: netip.MustParseAddr(networking.ToStandaloneAddress(address)), Port: port}, + OwnerProcess: otherOwner, + }) + require.ErrorIs(t, reserveErr, statestore.ErrPortReservationHeld) +} + +func requirePortReleased( + t *testing.T, + ctx context.Context, + store *statestore.Store, + protocol apiv1.PortProtocol, + address string, + port int32, + owner process.ProcessTreeItem, +) { + t.Helper() + + otherOwner := owner + otherOwner.IdentityTime = otherOwner.IdentityTime.Add(time.Second) + _, reserveErr := store.CreatePortReservation(ctx, statestore.PortReservationRequest{ + Binding: ports.Binding{Protocol: string(protocol), IP: netip.MustParseAddr(networking.ToStandaloneAddress(address)), Port: port}, + OwnerProcess: otherOwner, + }) + require.NoError(t, reserveErr) +} diff --git a/internal/dcpctrl/commands/run_controllers.go b/internal/dcpctrl/commands/run_controllers.go index 2ccafa42..b47a9a08 100644 --- a/internal/dcpctrl/commands/run_controllers.go +++ b/internal/dcpctrl/commands/run_controllers.go @@ -26,6 +26,7 @@ import ( dcptunproto "github.com/microsoft/dcp/internal/dcptun/proto" "github.com/microsoft/dcp/internal/exerunners" "github.com/microsoft/dcp/internal/health" + "github.com/microsoft/dcp/internal/networking" "github.com/microsoft/dcp/internal/notifications" "github.com/microsoft/dcp/internal/perftrace" "github.com/microsoft/dcp/internal/proxy" @@ -133,6 +134,10 @@ func runControllers(log logr.Logger) func(cmd *cobra.Command, _ []string) error if cleanupErr := stateStore.DeleteInactiveResourceLeases(ctrlCtx); cleanupErr != nil { log.Error(cleanupErr, "Failed to clean up inactive state store resource leases") } + if cleanupErr := stateStore.DeleteInactivePortReservations(ctrlCtx); cleanupErr != nil { + log.Error(cleanupErr, "Failed to clean up inactive state store port reservations") + } + networking.ConfigureStateStorePortAllocator(stateStore, leaseOwner) startInvalidPersistentExecutableRecordCleanup(ctrlCtx, stateStore, leaseOwner, log) trySetupNotificationHandler(ctrlCtx, log) diff --git a/internal/dcptun/tunnel_test.go b/internal/dcptun/tunnel_test.go index 77444df4..f876738e 100644 --- a/internal/dcptun/tunnel_test.go +++ b/internal/dcptun/tunnel_test.go @@ -170,7 +170,7 @@ func TestTunnelDataThroughput(t *testing.T) { defer testCtxCancel() log := testutil.NewLogForTesting(t.Name()) - serverPort, portErr := networking.GetFreePort(apiv1.TCP, networking.IPv4LocalhostDefaultAddress, log) + serverPort, portErr := networking.GetFreePort(context.Background(), apiv1.TCP, networking.IPv4LocalhostDefaultAddress, log) require.NoError(t, portErr, "Failed to get free port for server") serverProxyClient, proxyCleanup := createProxyPair(t, testCtx, log) diff --git a/internal/networking/networking.go b/internal/networking/networking.go index 734edeb6..29f5496c 100644 --- a/internal/networking/networking.go +++ b/internal/networking/networking.go @@ -9,20 +9,17 @@ import ( "context" "fmt" "net" + "net/netip" "os" - "path/filepath" - std_slices "slices" "strings" "sync" "time" - "github.com/go-logr/logr" "golang.org/x/net/nettest" - "k8s.io/apimachinery/pkg/util/wait" - apiv1 "github.com/microsoft/dcp/api/v1" - "github.com/microsoft/dcp/internal/dcppaths" - "github.com/microsoft/dcp/pkg/osutil" + "github.com/microsoft/dcp/internal/statestore" + "github.com/microsoft/dcp/pkg/ports" + "github.com/microsoft/dcp/pkg/process" "github.com/microsoft/dcp/pkg/randdata" "github.com/microsoft/dcp/pkg/slices" ) @@ -49,12 +46,25 @@ const ( NetworkOpTimeout = 2 * time.Second DCP_INSTANCE_ID_PREFIX = "DCP_INSTANCE_ID_PREFIX" + + DCP_PORT_ALLOCATOR = "DCP_PORT_ALLOCATOR" + DCP_PORT_ALLOCATION_RANGE = "DCP_PORT_ALLOCATION_RANGE" + DCP_PORT_ALLOCATION_ALLOW_EPHEMERAL_OVERLAP = "DCP_PORT_ALLOCATION_ALLOW_EPHEMERAL_OVERLAP" + + portAllocatorModeMru = "mru" + portAllocatorModeStateStore = "statestore" + portAllocatorModeStateStoreWithFallback = "statestore-with-mru-fallback" ) var ( - portFileLock *sync.Mutex + portFileLock *sync.Mutex + // portAllocatorLock protects quick package-level allocator configuration reads/writes. + portAllocatorLock *sync.Mutex portFileErrorReported bool + portRangeOverlapReported bool packageMruPortFile *mruPortFile + packageStateStore *statestore.Store + packageStateStoreOwner process.ProcessTreeItem programInstanceID string ipVersionPreference IpVersionPreference getAllLocalIpsOnce func() ([]net.IP, error) @@ -69,6 +79,7 @@ type portRange struct { func init() { portFileLock = &sync.Mutex{} + portAllocatorLock = &sync.Mutex{} getAllLocalIpsOnce = sync.OnceValues(getAllLocalIps) getHostnameOnce = sync.OnceValues(os.Hostname) getEphemeralPortRangeOnce = sync.OnceValues(getEphemeralPortRange) @@ -98,6 +109,14 @@ func init() { } } +func ConfigureStateStorePortAllocator(store *statestore.Store, owner process.ProcessTreeItem) { + portAllocatorLock.Lock() + defer portAllocatorLock.Unlock() + + packageStateStore = store + packageStateStoreOwner = owner +} + // Wrap the standard net.LookupIP method to filter for supported IP address types func LookupIP(host string) ([]net.IP, error) { var validIps []net.IP @@ -139,9 +158,9 @@ func LookupIP(host string) ([]net.IP, error) { validIps = slices.Select(ips, IsValidIP) default: - if ip := net.ParseIP(host); ip != nil { + if addr, parseErr := netip.ParseAddr(host); parseErr == nil { // This is a direct IP address, so we attempt to use it as is - validIps = slices.Select([]net.IP{ip}, IsValidIP) + validIps = slices.Select([]net.IP{netIPFromAddr(addr)}, IsValidIP) } else { // This is an arbitrary hostname, so resolve it to all local IP addresses allLocalIps, err := getAllLocalIpsOnce() @@ -220,160 +239,78 @@ func GetEphemeralPortRange() (int, int, bool) { return ephemeralRange.Start, ephemeralRange.End, matched } -// Gets a free TCP or UDP port for a given address (defaults to localhost). -// Even if this method is called twice in a row, it should not return the same port. -func GetFreePort(protocol apiv1.PortProtocol, address string, log logr.Logger) (int32, error) { - if address == "" { - address = Localhost - } - - if address != Localhost && net.ParseIP(ToStandaloneAddress(address)) == nil { - // We treat any non-Localhost and non-IP address as a request to bind to all interfaces. - address = IPv4AllInterfaceAddress - } - - portFileLock.Lock() - portFileErr := ensurePackageMruPortFile(log) - portFileLock.Unlock() - var allocatedPort int32 = 0 - - bestEffortPortAllocation := func(err error) (int32, error) { - if packageMruPortFile.params.failOnPortFileError { - return 0, err - } else { - return doGetFreePort(protocol, address) +// NormalizePortAllocationAddress converts a caller-provided bind address into the concrete address +// used by the port allocator. +func NormalizePortAllocationAddress(address string) (string, error) { + if address == "" || address == Localhost { + preferredIps, preferredErr := GetPreferredHostIps(Localhost) + if preferredErr != nil { + return "", preferredErr } - } - - if portFileErr != nil { - // Error will be logged by ensurePackageMruPortFile() - log.V(1).Info("Warning: port allocation check: could not check the most recently used ports file for conflicts") - return bestEffortPortAllocation(portFileErr) - } - - mruPortsAllocCtx, cancel := context.WithTimeout(context.Background(), packageMruPortFile.params.portAllocationTimeout) - defer cancel() - - pollWaitErr := wait.PollUntilContextCancel(mruPortsAllocCtx, packageMruPortFile.params.portAllocationRoundDelay, true /* poll immediately */, func(ctx context.Context) (bool, error) { - portFileLock.Lock() - defer portFileLock.Unlock() - - usedPorts, usedPortsErr := packageMruPortFile.tryLockAndRead(mruPortsAllocCtx) - if usedPortsErr != nil { - return false, usedPortsErr - } - defer func() { - if unlockErr := packageMruPortFile.Unlock(); unlockErr != nil { - log.Error(unlockErr, "Could not unlock the most recently used ports file") // Should never happen - if packageMruPortFile.params.failOnPortFileError { - panic(unlockErr) - } - } - }() - - for i := 0; i < packageMruPortFile.params.portAllocationsPerRound; i++ { - port, portErr := doGetFreePort(protocol, address) - if portErr != nil { - return false, portErr - } - - _, recentlyUsed := std_slices.BinarySearchFunc(usedPorts, AddressAndPort(address, port), matchAddressAndPort) - - if !recentlyUsed { - allocatedPort = port - usedPorts = append(usedPorts, mruPortFileRecord{ - Address: address, - Port: allocatedPort, - AddressAndPort: AddressAndPort(address, allocatedPort), - Timestamp: time.Now(), - Instance: programInstanceID, - }) - if writeErr := packageMruPortFile.WriteAndUnlock(ctx, usedPorts); writeErr != nil { - log.Error(writeErr, "Could not write to the most recently used ports file") - if packageMruPortFile.params.failOnPortFileError { - return false, writeErr - } - } - return true, nil - } + preferredAddr, ok := addrFromNetIP(preferredIps[0]) + if !ok { + return "", fmt.Errorf("could not parse preferred localhost IP address: %s", preferredIps[0].String()) } - - return false, nil // Keep trying - }) - - if pollWaitErr == nil { - return allocatedPort, nil - } else { - log.V(1).Info("Warning: port allocation check: could not check the most recently used ports file for conflicts", - "Error", pollWaitErr.Error()) - return bestEffortPortAllocation(pollWaitErr) - } -} - -func CheckPortAvailable(protocol apiv1.PortProtocol, address string, port int32, log logr.Logger) error { - if address == "" { - address = Localhost + return addrToAddressString(preferredAddr), nil } - - portFileLock.Lock() - defer portFileLock.Unlock() - - portFileErr := ensurePackageMruPortFile(log) - bestEffortCheck := func(err error) error { - if packageMruPortFile.params.failOnPortFileError { - return err - } else { - // Do best-effort check without considering the most recently used ports - return doCheckPortAvailable(protocol, address, port) - } + addr, parseErr := netip.ParseAddr(ToStandaloneAddress(address)) + if parseErr != nil { + // We treat any non-Localhost and non-IP address as a request to bind to all interfaces. + return IPv4AllInterfaceAddress, nil } + return addrToAddressString(addr), nil +} - if portFileErr != nil { - // Error will be logged by ensurePackageMruPortFile() - return bestEffortCheck(portFileErr) +func IpToString(ip net.IP) string { + addr, ok := addrFromNetIP(ip) + if !ok { + return ip.String() } + return addrToAddressString(addr) +} - // Check if the port was allocated by one of the DCP instances - mruPortsCheckCtx, cancel := context.WithTimeout(context.Background(), packageMruPortFile.params.portAvailableCheckTimeout) - defer cancel() +func IsIPv4(address string) bool { + addr, parseErr := netip.ParseAddr(ToStandaloneAddress(address)) + return parseErr == nil && addr.Unmap().Is4() && nettest.SupportsIPv4() +} - usedPorts, usedPortsErr := packageMruPortFile.tryLockAndRead(mruPortsCheckCtx) - if usedPortsErr != nil { - log.V(1).Info("Warning: port availability check: could not check the most recently used ports file for conflicts", "Error", usedPortsErr.Error()) - return bestEffortCheck(usedPortsErr) - } +func IsIPv6(address string) bool { + addr, parseErr := netip.ParseAddr(ToStandaloneAddress(address)) + return parseErr == nil && addr.Is6() && !addr.Is4In6() && nettest.SupportsIPv6() +} - // Since we are not going to write anything to the file, we can unlock it immediately - unlockErr := packageMruPortFile.Unlock() - if unlockErr != nil { - log.Error(unlockErr, "Could not unlock the most recently used ports file") // Should never happen +func addrFromNetIP(ip net.IP) (netip.Addr, bool) { + ip4 := ip.To4() + if ip4 != nil { + var bytes [4]byte + copy(bytes[:], ip4) + return netip.AddrFrom4(bytes), true } - - desired := AddressAndPort(address, port) - i, found := std_slices.BinarySearchFunc(usedPorts, desired, matchAddressAndPort) - if found && usedPorts[i].Instance != programInstanceID { - return fmt.Errorf("port %d is already in use by another process", port) - } else { - return doCheckPortAvailable(protocol, address, port) + ip16 := ip.To16() + if ip16 == nil { + return netip.Addr{}, false } + var bytes [16]byte + copy(bytes[:], ip16) + return netip.AddrFrom16(bytes), true } -func IpToString(ip net.IP) string { - var address string - if IsValidIPv6(ip) { - address = fmt.Sprintf("[%s]", ip.String()) - } else { - address = ip.String() +func netIPFromAddr(addr netip.Addr) net.IP { + addr = addr.Unmap() + if addr.Is4() { + bytes := addr.As4() + return net.IPv4(bytes[0], bytes[1], bytes[2], bytes[3]) } - return address + bytes := addr.As16() + return append(net.IP(nil), bytes[:]...) } -func IsIPv4(address string) bool { - return IsValidIPv4(net.ParseIP(ToStandaloneAddress(address))) -} - -func IsIPv6(address string) bool { - return IsValidIPv6(net.ParseIP(ToStandaloneAddress(address))) +func addrToAddressString(addr netip.Addr) string { + addr = addr.Unmap() + if addr.Is6() { + return fmt.Sprintf("[%s]", addr.String()) + } + return addr.String() } // Removes the brackets from an IPv6 address if they are present. @@ -390,11 +327,11 @@ func ToStandaloneAddress(address string) string { } func IsValidPort(port int) bool { - return port >= 1 && port <= 65535 + return ports.IsValidPort(port) } func IsBindablePort(port int) bool { - return port == 0 || IsValidPort(port) + return ports.IsBindablePort(port) } func IsEphemeralPort(port int32) bool { @@ -422,15 +359,6 @@ func IsValidIPv6(ip net.IP) bool { return len(ip.To16()) == net.IPv6len && len(ip.To4()) == 0 && nettest.SupportsIPv6() } -// Used by tests, makes the use of the most recently used ports file mandatory -func EnableStrictMruPortHandling(log logr.Logger) { - err := ensurePackageMruPortFile(log) - if err != nil { - panic("could not create the most recently used ports file; network port conflicts may occur") - } - packageMruPortFile.params.failOnPortFileError = true -} - func GetProgramInstanceID() string { return programInstanceID } @@ -439,64 +367,6 @@ func GetIpVersionPreference() IpVersionPreference { return ipVersionPreference } -func doCheckPortAvailable(protocol apiv1.PortProtocol, address string, port int32) error { - if protocol == apiv1.UDP { - udpaddr, err := net.ResolveUDPAddr("udp", AddressAndPort(address, port)) - if err != nil { - return err - } - - if listener, listenErr := net.ListenUDP("udp", udpaddr); listenErr != nil { - return listenErr - } else { - listener.Close() - return nil - } - } else { - tcpaddr, err := net.ResolveTCPAddr("tcp", AddressAndPort(address, port)) - if err != nil { - return err - } - - if listener, listenErr := net.ListenTCP("tcp", tcpaddr); listenErr != nil { - return listenErr - } else { - listener.Close() - return nil - } - } -} - -func doGetFreePort(protocol apiv1.PortProtocol, address string) (int32, error) { - if protocol == apiv1.UDP { - udpaddr, err := net.ResolveUDPAddr("udp", AddressAndPort(address, 0)) - if err != nil { - return 0, err - } - - if listener, listenErr := net.ListenUDP("udp", udpaddr); listenErr != nil { - return 0, listenErr - } else { - port := int32(listener.LocalAddr().(*net.UDPAddr).Port) - listener.Close() - return port, nil - } - } else { - tcpaddr, err := net.ResolveTCPAddr("tcp", AddressAndPort(address, 0)) - if err != nil { - return 0, err - } - - if listener, listenErr := net.ListenTCP("tcp", tcpaddr); listenErr != nil { - return 0, listenErr - } else { - port := int32(listener.Addr().(*net.TCPAddr).Port) - listener.Close() - return port, nil - } - } -} - func AddressAndPort(address string, port int32) string { return fmt.Sprintf("%s:%d", address, port) } @@ -507,44 +377,6 @@ func Hostname() (string, error) { return getHostnameOnce() } -// Creates the (default) most-recently-used ports file. -// Assumes mruPortFileLock is held (this function is not goroutine-safe). -func ensurePackageMruPortFile(log logr.Logger) error { - if packageMruPortFile != nil { - return nil - } - - err := func() error { - dcpFolder, dcpFolderErr := dcppaths.EnsureUserDcpDir() - if dcpFolderErr != nil { - return dcpFolderErr - } - - isAdmin, isAdminErr := osutil.IsAdmin() - if isAdminErr != nil { - return isAdminErr - } - - var lockfilePath string - if isAdmin { - lockfilePath = filepath.Join(dcpFolder, "mruPorts.elevated.list") - } else { - lockfilePath = filepath.Join(dcpFolder, "mruPorts.list") - } - - var creationErr error - packageMruPortFile, creationErr = newMruPortFile(lockfilePath, defaultMruPortFileUsageParameters()) - return creationErr - }() - - if err != nil && !portFileErrorReported { - portFileErrorReported = true - log.Error(err, "Could not create the most recently used ports file; network port conflicts may occur") - } - - return err -} - func GetPreferredHostIps(host string) ([]net.IP, error) { // Some host names are pseudo-addresses that resolve to multiple IP addresses. // For example, "localhost" hostname resolves to all loopback addresses on the machine. For dual-stack machines (very common) diff --git a/internal/networking/networking_test.go b/internal/networking/networking_test.go index f7c82017..e6c16d57 100644 --- a/internal/networking/networking_test.go +++ b/internal/networking/networking_test.go @@ -6,14 +6,24 @@ package networking import ( + "context" + "fmt" "io" "net" + "net/netip" "os" + "path/filepath" "sync" "testing" + "time" apiv1 "github.com/microsoft/dcp/api/v1" + "github.com/microsoft/dcp/internal/statestore" + usvc_io "github.com/microsoft/dcp/pkg/io" "github.com/microsoft/dcp/pkg/logger" + "github.com/microsoft/dcp/pkg/osutil" + "github.com/microsoft/dcp/pkg/process" + "github.com/microsoft/dcp/pkg/testutil" "github.com/stretchr/testify/require" ) @@ -52,8 +62,8 @@ func TestGetFreePort(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() - port1, err1 := GetFreePort(tc.protocol, Localhost, log) - port2, err2 := GetFreePort(tc.protocol, Localhost, log) + port1, err1 := GetFreePort(context.Background(), tc.protocol, Localhost, log) + port2, err2 := GetFreePort(context.Background(), tc.protocol, Localhost, log) require.NoError(t, err1, "error1: %v", err1) require.NoError(t, err2, "error2: %v", err2) @@ -77,7 +87,7 @@ func TestCanGetFreePortForAllLocalIPs(t *testing.T) { for _, ip := range ips { address := IpToString(ip) - _, err = GetFreePort(tc.protocol, address, log) + _, err = GetFreePort(context.Background(), tc.protocol, address, log) require.NoError(t, err, "Could not get free port for address %s", address) } }) @@ -104,10 +114,10 @@ func TestCheckPortAvailable(t *testing.T) { go func(ip net.IP) { defer wg.Done() address := IpToString(ip) - port, err := GetFreePort(tc.protocol, address, log) + port, err := GetFreePort(context.Background(), tc.protocol, address, log) require.NoError(t, err, "Could not get free port for address %s", address) - err = CheckPortAvailable(tc.protocol, address, port, log) + err = CheckPortAvailable(context.Background(), tc.protocol, address, port, log) require.NoError(t, err, "Port %d on address %s is not available", port, address) // Occupy the port @@ -126,13 +136,13 @@ func TestCheckPortAvailable(t *testing.T) { require.NoError(t, err, "Could not listen on TCP address %s", ap) } - err = CheckPortAvailable(tc.protocol, address, port, log) + err = CheckPortAvailable(context.Background(), tc.protocol, address, port, log) require.Error(t, err, "Port %d on address %s is available", port, address) err = listener.Close() require.NoError(t, err, "Could not close listener on port %d on address %s", port, address) - err = CheckPortAvailable(tc.protocol, address, port, log) + err = CheckPortAvailable(context.Background(), tc.protocol, address, port, log) require.NoError(t, err, "Port %d on address %s is not available", port, address) }(ip) } @@ -141,3 +151,170 @@ func TestCheckPortAvailable(t *testing.T) { }) } } + +func TestConfiguredPortAllocationRangesSubtractsEphemeralOverlap(t *testing.T) { + ephemeralStart, ephemeralEnd, _ := GetEphemeralPortRange() + if ephemeralStart <= 1 { + t.Skip("ephemeral range starts too low to create a partially overlapping test range") + } + + t.Setenv(DCP_PORT_ALLOCATION_RANGE, fmt.Sprintf("%d-%d", ephemeralStart-1, ephemeralStart+1)) + portRangeOverlapReported = false + + ranges, err := configuredPortAllocationRanges(false, log) + + require.NoError(t, err) + for _, candidateRange := range ranges.ranges { + require.True(t, candidateRange.End < ephemeralStart || candidateRange.Start > ephemeralEnd) + } +} + +func TestIndexedPortRangesNormalizeOverlappingRanges(t *testing.T) { + t.Parallel() + + ranges := newIndexedPortRanges([]portRange{ + {Start: 26020, End: 26022}, + {Start: 26010, End: 26015}, + {Start: 26014, End: 26018}, + {Start: 26030, End: 26030}, + {Start: 26019, End: 26025}, + }) + + require.Equal(t, 17, ranges.Len()) + require.Equal(t, []portRange{ + {Start: 26010, End: 26025}, + {Start: 26030, End: 26030}, + }, ranges.ranges) + + candidates := []int32{} + for index := 0; index < ranges.Len(); index++ { + candidate, ok := ranges.PortAt(index) + require.True(t, ok) + candidates = append(candidates, candidate) + } + + require.Equal(t, []int32{ + 26010, 26011, 26012, 26013, 26014, 26015, 26016, 26017, 26018, + 26019, 26020, 26021, 26022, 26023, 26024, 26025, 26030, + }, candidates) + + _, beforeRangeOk := ranges.PortAt(-1) + _, afterRangeOk := ranges.PortAt(ranges.Len()) + require.False(t, beforeRangeOk) + require.False(t, afterRangeOk) +} + +func TestPortAllocationCandidatesCoverRange(t *testing.T) { + t.Parallel() + + ranges := newIndexedPortRanges([]portRange{ + {Start: 26010, End: 26011}, + {Start: 26020, End: 26022}, + }) + + candidateIterator := newStateStorePortAllocationCandidateIterator(ranges, string(apiv1.TCP), IPv4LocalhostDefaultAddress) + candidates := []int32{} + for { + candidate, ok := candidateIterator.Next() + if !ok { + break + } + candidates = append(candidates, candidate) + } + + require.Len(t, candidates, 5) + require.ElementsMatch(t, []int32{26010, 26011, 26020, 26021, 26022}, candidates) +} + +func TestPortAllocationCandidatesCoverOverlappingRangesOnce(t *testing.T) { + t.Parallel() + + ranges := newIndexedPortRanges([]portRange{ + {Start: 26010, End: 26015}, + {Start: 26013, End: 26017}, + {Start: 26020, End: 26021}, + }) + + candidateIterator := newStateStorePortAllocationCandidateIterator(ranges, string(apiv1.TCP), IPv4LocalhostDefaultAddress) + candidates := []int32{} + for { + candidate, ok := candidateIterator.Next() + if !ok { + break + } + candidates = append(candidates, candidate) + } + + require.Len(t, candidates, 10) + require.ElementsMatch(t, []int32{26010, 26011, 26012, 26013, 26014, 26015, 26016, 26017, 26020, 26021}, candidates) +} + +func TestPortAllocationCandidateStepCoversEveryIndex(t *testing.T) { + t.Parallel() + + for total := 2; total <= 100; total++ { + step := stateStorePortAllocationCandidateStep(total, uint64(total*17)) + seen := map[int]struct{}{} + for next := 0; next < total; next++ { + seen[(next*step)%total] = struct{}{} + } + + require.Len(t, seen, total) + } +} + +func TestNormalizePortAllocationAddressUsesIPForLocalhost(t *testing.T) { + t.Parallel() + + address, err := NormalizePortAllocationAddress(Localhost) + + require.NoError(t, err) + require.NotEqual(t, Localhost, address) + _, parseErr := netip.ParseAddr(ToStandaloneAddress(address)) + require.NoError(t, parseErr) +} + +func TestNormalizePortAllocationAddressCanonicalizesIPs(t *testing.T) { + t.Parallel() + + ipv4MappedAddress, ipv4MappedErr := NormalizePortAllocationAddress("[::ffff:127.0.0.1]") + require.NoError(t, ipv4MappedErr) + require.Equal(t, IPv4LocalhostDefaultAddress, ipv4MappedAddress) + + ipv6Address, ipv6Err := NormalizePortAllocationAddress("[0:0:0:0:0:0:0:1]") + require.NoError(t, ipv6Err) + require.Equal(t, IPv6LocalhostDefaultAddress, ipv6Address) +} + +func TestGetFreePortWithStateStoreUsesConfiguredRange(t *testing.T) { + ctx, cancel := testutil.GetTestContext(t, 10*time.Second) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + require.NoError(t, usvc_io.EnsureRestrictedDirectory(filepath.Dir(storePath), osutil.PermissionOnlyOwnerReadWriteTraverse)) + store, openErr := statestore.Open(ctx, statestore.Options{ + Path: storePath, + BusyTimeout: 500 * time.Millisecond, + }) + require.NoError(t, openErr) + defer func() { + require.NoError(t, store.Close()) + }() + + owner, ownerErr := process.This() + require.NoError(t, ownerErr) + ConfigureStateStorePortAllocator(store, owner) + defer ConfigureStateStorePortAllocator(nil, process.ProcessTreeItem{}) + t.Setenv(DCP_PORT_ALLOCATOR, portAllocatorModeStateStore) + t.Setenv(DCP_PORT_ALLOCATION_RANGE, "26010-26020") + + port1, port1Err := GetFreePort(ctx, apiv1.TCP, IPv4LocalhostDefaultAddress, log) + port2, port2Err := GetFreePort(ctx, apiv1.TCP, IPv4LocalhostDefaultAddress, log) + + require.NoError(t, port1Err) + require.NoError(t, port2Err) + require.NotEqual(t, port1, port2) + require.GreaterOrEqual(t, port1, int32(26010)) + require.LessOrEqual(t, port1, int32(26020)) + require.GreaterOrEqual(t, port2, int32(26010)) + require.LessOrEqual(t, port2, int32(26020)) +} diff --git a/internal/networking/port_allocator.go b/internal/networking/port_allocator.go new file mode 100644 index 00000000..63838c00 --- /dev/null +++ b/internal/networking/port_allocator.go @@ -0,0 +1,79 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package networking + +import ( + "context" + + "github.com/go-logr/logr" + + apiv1 "github.com/microsoft/dcp/api/v1" + "github.com/microsoft/dcp/pkg/ports" +) + +// Gets a free TCP or UDP port for a given address (defaults to localhost). +// Even if this method is called twice in a row, it should not return the same port. +func GetFreePort(ctx context.Context, protocol apiv1.PortProtocol, address string, log logr.Logger) (int32, error) { + address, addressErr := NormalizePortAllocationAddress(address) + if addressErr != nil { + return 0, addressErr + } + + stateStorePort, stateStoreErr, shouldFallback := allocatePortFromStateStoreRange(ctx, protocol, address, log) + if stateStoreErr == nil { + return stateStorePort, nil + } + if !shouldFallback { + return 0, stateStoreErr + } + log.V(1).Info("Warning: state store port allocation failed, falling back to MRU port allocation", "Error", stateStoreErr.Error()) + return allocateRandomEphemeralPortWithMruFile(ctx, protocol, address, log) +} + +func CheckPortAvailable(ctx context.Context, protocol apiv1.PortProtocol, address string, port int32, log logr.Logger) error { + address, addressErr := NormalizePortAllocationAddress(address) + if addressErr != nil { + return addressErr + } + + stateStoreErr, shouldFallback := checkPortAvailableWithStateStore(ctx, protocol, address, port, log) + if stateStoreErr == nil { + return nil + } + if !shouldFallback { + return stateStoreErr + } + if stateStoreErr != nil { + log.V(1).Info("Warning: state store port availability check failed, falling back to MRU port availability check", "Error", stateStoreErr.Error()) + } + return checkPortAvailableWithMruFile(ctx, protocol, address, port, log) +} + +func ReserveSpecificPort(ctx context.Context, binding ports.Binding, log logr.Logger) error { + stateStoreErr, shouldFallback := reserveSpecificPortWithStateStore(ctx, binding, log) + if stateStoreErr == nil { + return nil + } + if !shouldFallback { + return stateStoreErr + } + if stateStoreErr != nil { + log.V(1).Info("Warning: state store port reservation failed, continuing without MRU reservation", "Error", stateStoreErr.Error()) + } + return nil +} + +func ReleaseSpecificPort(ctx context.Context, binding ports.Binding, log logr.Logger) error { + stateStoreErr, shouldFallback := releaseSpecificPortWithStateStore(ctx, binding, log) + if stateStoreErr == nil { + return nil + } + if !shouldFallback { + return stateStoreErr + } + log.V(1).Info("Warning: state store port release failed, falling back to MRU port release", "Error", stateStoreErr.Error()) + return releaseSpecificPortWithMruFile(ctx, addrToAddressString(binding.IP), binding.Port, log) +} diff --git a/internal/networking/port_allocator_mru.go b/internal/networking/port_allocator_mru.go new file mode 100644 index 00000000..330f6fd6 --- /dev/null +++ b/internal/networking/port_allocator_mru.go @@ -0,0 +1,308 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package networking + +import ( + "context" + "fmt" + "net" + "path/filepath" + std_slices "slices" + "time" + + "github.com/go-logr/logr" + "k8s.io/apimachinery/pkg/util/wait" + + apiv1 "github.com/microsoft/dcp/api/v1" + "github.com/microsoft/dcp/internal/dcppaths" + "github.com/microsoft/dcp/pkg/osutil" +) + +func allocateRandomEphemeralPortWithMruFile(ctx context.Context, protocol apiv1.PortProtocol, address string, log logr.Logger) (int32, error) { + portFileLock.Lock() + portFileErr := ensurePackageMruPortFile(log) + portFileLock.Unlock() + var allocatedPort int32 = 0 + + bestEffortPortAllocation := func(err error) (int32, error) { + if packageMruPortFile.params.failOnPortFileError { + return 0, err + } else { + return getRandomEphemeralPort(protocol, address) + } + } + + if portFileErr != nil { + // Error will be logged by ensurePackageMruPortFile() + log.V(1).Info("Warning: port allocation check: could not check the most recently used ports file for conflicts") + return bestEffortPortAllocation(portFileErr) + } + + mruPortsAllocCtx, cancel := context.WithTimeout(ctx, packageMruPortFile.params.portAllocationTimeout) + defer cancel() + + pollWaitErr := wait.PollUntilContextCancel(mruPortsAllocCtx, packageMruPortFile.params.portAllocationRoundDelay, true /* poll immediately */, func(ctx context.Context) (bool, error) { + portFileLock.Lock() + defer portFileLock.Unlock() + + usedPorts, usedPortsErr := packageMruPortFile.tryLockAndRead(mruPortsAllocCtx) + if usedPortsErr != nil { + return false, usedPortsErr + } + defer func() { + if unlockErr := packageMruPortFile.Unlock(); unlockErr != nil { + log.Error(unlockErr, "Could not unlock the most recently used ports file") // Should never happen + if packageMruPortFile.params.failOnPortFileError { + panic(unlockErr) + } + } + }() + + for i := 0; i < packageMruPortFile.params.portAllocationsPerRound; i++ { + port, portErr := getRandomEphemeralPort(protocol, address) + if portErr != nil { + return false, portErr + } + + _, recentlyUsed := std_slices.BinarySearchFunc(usedPorts, AddressAndPort(address, port), matchAddressAndPort) + + if !recentlyUsed { + allocatedPort = port + usedPorts = append(usedPorts, mruPortFileRecord{ + Address: address, + Port: allocatedPort, + AddressAndPort: AddressAndPort(address, allocatedPort), + Timestamp: time.Now(), + Instance: programInstanceID, + }) + if writeErr := packageMruPortFile.WriteAndUnlock(ctx, usedPorts); writeErr != nil { + log.Error(writeErr, "Could not write to the most recently used ports file") + if packageMruPortFile.params.failOnPortFileError { + return false, writeErr + } + } + return true, nil + } + } + + return false, nil // Keep trying + }) + + if pollWaitErr == nil { + return allocatedPort, nil + } else { + log.V(1).Info("Warning: port allocation check: could not check the most recently used ports file for conflicts", + "Error", pollWaitErr.Error()) + return bestEffortPortAllocation(pollWaitErr) + } +} + +func checkPortAvailableWithMruFile(ctx context.Context, protocol apiv1.PortProtocol, address string, port int32, log logr.Logger) error { + portFileLock.Lock() + defer portFileLock.Unlock() + + portFileErr := ensurePackageMruPortFile(log) + bestEffortCheck := func(err error) error { + if packageMruPortFile.params.failOnPortFileError { + return err + } else { + // Do best-effort check without considering the most recently used ports + return checkPortCurrentlyBindable(protocol, address, port) + } + } + + if portFileErr != nil { + // Error will be logged by ensurePackageMruPortFile() + return bestEffortCheck(portFileErr) + } + + // Check if the port was allocated by one of the DCP instances + mruPortsCheckCtx, cancel := context.WithTimeout(ctx, packageMruPortFile.params.portAvailableCheckTimeout) + defer cancel() + + usedPorts, usedPortsErr := packageMruPortFile.tryLockAndRead(mruPortsCheckCtx) + if usedPortsErr != nil { + log.V(1).Info("Warning: port availability check: could not check the most recently used ports file for conflicts", "Error", usedPortsErr.Error()) + return bestEffortCheck(usedPortsErr) + } + + // Since we are not going to write anything to the file, we can unlock it immediately + unlockErr := packageMruPortFile.Unlock() + if unlockErr != nil { + log.Error(unlockErr, "Could not unlock the most recently used ports file") // Should never happen + } + + desired := AddressAndPort(address, port) + i, found := std_slices.BinarySearchFunc(usedPorts, desired, matchAddressAndPort) + if found && usedPorts[i].Instance != programInstanceID { + return fmt.Errorf("port %d is already in use by another process", port) + } else { + return checkPortCurrentlyBindable(protocol, address, port) + } +} + +func releaseSpecificPortWithMruFile(ctx context.Context, address string, port int32, log logr.Logger) error { + portFileLock.Lock() + defer portFileLock.Unlock() + + portFileErr := ensurePackageMruPortFile(log) + params := defaultMruPortFileUsageParameters() + if packageMruPortFile != nil { + params = packageMruPortFile.params + } + bestEffortRelease := func(err error) error { + if params.failOnPortFileError { + return err + } + return nil + } + + if portFileErr != nil { + // Error will be logged by ensurePackageMruPortFile() + return bestEffortRelease(portFileErr) + } + + mruPortsReleaseCtx, cancel := context.WithTimeout(ctx, packageMruPortFile.params.portAvailableCheckTimeout) + defer cancel() + + usedPorts, usedPortsErr := packageMruPortFile.tryLockAndRead(mruPortsReleaseCtx) + if usedPortsErr != nil { + log.V(1).Info("Warning: port release: could not check the most recently used ports file for conflicts", "Error", usedPortsErr.Error()) + return bestEffortRelease(usedPortsErr) + } + + desired := AddressAndPort(address, port) + filteredPorts := usedPorts[:0] + released := false + for _, usedPort := range usedPorts { + if usedPort.AddressAndPort == desired && usedPort.Instance == programInstanceID { + released = true + continue + } + filteredPorts = append(filteredPorts, usedPort) + } + + if !released { + unlockErr := packageMruPortFile.Unlock() + if unlockErr != nil { + log.Error(unlockErr, "Could not unlock the most recently used ports file") + return bestEffortRelease(unlockErr) + } + return nil + } + + writeErr := packageMruPortFile.WriteAndUnlock(mruPortsReleaseCtx, filteredPorts) + if writeErr != nil { + log.Error(writeErr, "Could not write to the most recently used ports file") + return bestEffortRelease(writeErr) + } + return nil +} + +func checkPortCurrentlyBindable(protocol apiv1.PortProtocol, address string, port int32) error { + if protocol == apiv1.UDP { + udpaddr, err := net.ResolveUDPAddr("udp", AddressAndPort(address, port)) + if err != nil { + return err + } + + if listener, listenErr := net.ListenUDP("udp", udpaddr); listenErr != nil { + return listenErr + } else { + listener.Close() + return nil + } + } else { + tcpaddr, err := net.ResolveTCPAddr("tcp", AddressAndPort(address, port)) + if err != nil { + return err + } + + if listener, listenErr := net.ListenTCP("tcp", tcpaddr); listenErr != nil { + return listenErr + } else { + listener.Close() + return nil + } + } +} + +func getRandomEphemeralPort(protocol apiv1.PortProtocol, address string) (int32, error) { + if protocol == apiv1.UDP { + udpaddr, err := net.ResolveUDPAddr("udp", AddressAndPort(address, 0)) + if err != nil { + return 0, err + } + + if listener, listenErr := net.ListenUDP("udp", udpaddr); listenErr != nil { + return 0, listenErr + } else { + port := int32(listener.LocalAddr().(*net.UDPAddr).Port) + listener.Close() + return port, nil + } + } else { + tcpaddr, err := net.ResolveTCPAddr("tcp", AddressAndPort(address, 0)) + if err != nil { + return 0, err + } + + if listener, listenErr := net.ListenTCP("tcp", tcpaddr); listenErr != nil { + return 0, listenErr + } else { + port := int32(listener.Addr().(*net.TCPAddr).Port) + listener.Close() + return port, nil + } + } +} + +// Used by tests, makes the use of the most recently used ports file mandatory. +func EnableStrictMruPortHandling(log logr.Logger) { + err := ensurePackageMruPortFile(log) + if err != nil { + panic("could not create the most recently used ports file; network port conflicts may occur") + } + packageMruPortFile.params.failOnPortFileError = true +} + +// Creates the (default) most-recently-used ports file. +// Assumes mruPortFileLock is held (this function is not goroutine-safe). +func ensurePackageMruPortFile(log logr.Logger) error { + if packageMruPortFile != nil { + return nil + } + + err := func() error { + dcpFolder, dcpFolderErr := dcppaths.EnsureUserDcpDir() + if dcpFolderErr != nil { + return dcpFolderErr + } + + isAdmin, isAdminErr := osutil.IsAdmin() + if isAdminErr != nil { + return isAdminErr + } + + var lockfilePath string + if isAdmin { + lockfilePath = filepath.Join(dcpFolder, "mruPorts.elevated.list") + } else { + lockfilePath = filepath.Join(dcpFolder, "mruPorts.list") + } + + var creationErr error + packageMruPortFile, creationErr = newMruPortFile(lockfilePath, defaultMruPortFileUsageParameters()) + return creationErr + }() + + if err != nil && !portFileErrorReported { + portFileErrorReported = true + log.Error(err, "Could not create the most recently used ports file; network port conflicts may occur") + } + + return err +} diff --git a/internal/networking/port_allocator_statestore.go b/internal/networking/port_allocator_statestore.go new file mode 100644 index 00000000..232ac0c8 --- /dev/null +++ b/internal/networking/port_allocator_statestore.go @@ -0,0 +1,468 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package networking + +import ( + "context" + "errors" + "fmt" + "hash/fnv" + "net/netip" + "os" + "sort" + "strconv" + "strings" + "time" + + "github.com/go-logr/logr" + + apiv1 "github.com/microsoft/dcp/api/v1" + "github.com/microsoft/dcp/internal/statestore" + "github.com/microsoft/dcp/pkg/concurrency" + "github.com/microsoft/dcp/pkg/osutil" + "github.com/microsoft/dcp/pkg/ports" + "github.com/microsoft/dcp/pkg/process" +) + +type stateStorePortAllocationConfig struct { + store *statestore.Store + owner process.ProcessTreeItem + mode string + portRanges indexedPortRanges + allowEphemeralOverlap bool + allocationTimeout time.Duration +} + +const ( + defaultStateStorePortAllocationRangeStart = 20000 + defaultStateStorePortAllocationRangeEnd = 32767 +) + +// stateStorePortReservationLock serializes slower bind probes with state-store reservation writes. +// Do not hold portAllocatorLock while acquiring stateStorePortReservationLock, or vice versa. +var stateStorePortReservationLock = concurrency.NewContextAwareLock() + +func getStateStorePortAllocationConfig(log logr.Logger) (*stateStorePortAllocationConfig, bool, error) { + mode := strings.ToLower(strings.TrimSpace(os.Getenv(DCP_PORT_ALLOCATOR))) + if mode == "" { + mode = portAllocatorModeStateStoreWithFallback + } + switch mode { + case portAllocatorModeMru: + return nil, true, fmt.Errorf("state store port allocation is disabled") + case portAllocatorModeStateStore, portAllocatorModeStateStoreWithFallback: + default: + return nil, false, fmt.Errorf("unsupported port allocator mode '%s'", mode) + } + + portAllocatorLock.Lock() + store := packageStateStore + owner := packageStateStoreOwner + portAllocatorLock.Unlock() + + if store == nil { + return nil, mode == portAllocatorModeStateStoreWithFallback, fmt.Errorf("state store port allocator is not configured") + } + + params := defaultMruPortFileUsageParameters() + allowEphemeralOverlap := osutil.EnvVarSwitchEnabled(DCP_PORT_ALLOCATION_ALLOW_EPHEMERAL_OVERLAP) + ranges, rangesErr := configuredPortAllocationRanges(allowEphemeralOverlap, log) + if rangesErr != nil { + return nil, mode == portAllocatorModeStateStoreWithFallback, rangesErr + } + + return &stateStorePortAllocationConfig{ + store: store, + owner: owner, + mode: mode, + portRanges: ranges, + allowEphemeralOverlap: allowEphemeralOverlap, + allocationTimeout: params.portAllocationTimeout, + }, false, nil +} + +func allocatePortFromStateStoreRange(ctx context.Context, protocol apiv1.PortProtocol, address string, log logr.Logger) (int32, error, bool) { + config, shouldFallback, configErr := getStateStorePortAllocationConfig(log) + if configErr != nil { + return 0, configErr, shouldFallback + } + + allocationCtx, cancel := context.WithTimeout(ctx, config.allocationTimeout) + defer cancel() + + candidates := newStateStorePortAllocationCandidateIterator(config.portRanges, string(protocol), address) + for { + if ctxErr := allocationCtx.Err(); ctxErr != nil { + return 0, ctxErr, config.mode == portAllocatorModeStateStoreWithFallback + } + candidate, hasCandidate := candidates.Next() + if !hasCandidate { + break + } + portAvailable, reserveErr := reserveStateStoreCandidateIfBindable(allocationCtx, config, protocol, address, candidate) + if !portAvailable { + continue + } + if reserveErr == nil { + return candidate, nil, false + } + if errors.Is(reserveErr, statestore.ErrPortReservationHeld) { + continue + } + return 0, reserveErr, config.mode == portAllocatorModeStateStoreWithFallback + } + + return 0, fmt.Errorf("could not find an available port in configured DCP port allocation range"), config.mode == portAllocatorModeStateStoreWithFallback +} + +func reserveStateStoreCandidateIfBindable( + ctx context.Context, + config *stateStorePortAllocationConfig, + protocol apiv1.PortProtocol, + address string, + port int32, +) (bool, error) { + if lockErr := stateStorePortReservationLock.Lock(ctx); lockErr != nil { + return true, lockErr + } + defer stateStorePortReservationLock.Unlock() + + if checkErr := checkPortCurrentlyBindable(protocol, address, port); checkErr != nil { + return false, nil + } + + normalizedAddress, addressErr := NormalizePortAllocationAddress(address) + if addressErr != nil { + return true, addressErr + } + ip, ipErr := netip.ParseAddr(ToStandaloneAddress(normalizedAddress)) + if ipErr != nil { + return true, fmt.Errorf("could not parse port reservation IP '%s': %w", normalizedAddress, ipErr) + } + request := stateStorePortReservationRequest(ports.Binding{Protocol: string(protocol), IP: ip, Port: port}, config.owner) + _, reserveErr := config.store.CreatePortReservation(ctx, request) + return true, reserveErr +} + +func checkPortAvailableWithStateStore(ctx context.Context, protocol apiv1.PortProtocol, address string, port int32, log logr.Logger) (error, bool) { + normalizedAddress, addressErr := NormalizePortAllocationAddress(address) + if addressErr != nil { + return addressErr, false + } + ip, ipErr := netip.ParseAddr(ToStandaloneAddress(normalizedAddress)) + if ipErr != nil { + return fmt.Errorf("could not parse port reservation IP '%s': %w", normalizedAddress, ipErr), false + } + return reserveSpecificStateStorePort(ctx, ports.Binding{Protocol: string(protocol), IP: ip, Port: port}, address, true, log) +} + +func reserveSpecificPortWithStateStore(ctx context.Context, binding ports.Binding, log logr.Logger) (error, bool) { + return reserveSpecificStateStorePort(ctx, binding, "", false, log) +} + +func reserveSpecificStateStorePort(ctx context.Context, binding ports.Binding, bindAddress string, requireBindable bool, log logr.Logger) (error, bool) { + config, shouldFallback, configErr := getStateStorePortAllocationConfig(log) + if configErr != nil { + return configErr, shouldFallback + } + + if IsEphemeralPort(binding.Port) && !config.allowEphemeralOverlap { + return nil, false + } + + reservationCtx, cancel := context.WithTimeout(ctx, config.allocationTimeout) + defer cancel() + + if lockErr := stateStorePortReservationLock.Lock(reservationCtx); lockErr != nil { + return lockErr, config.mode == portAllocatorModeStateStoreWithFallback + } + defer stateStorePortReservationLock.Unlock() + + if requireBindable { + checkErr := checkPortCurrentlyBindable(apiv1.PortProtocol(binding.Protocol), bindAddress, binding.Port) + if checkErr != nil { + return checkErr, false + } + } + + request := stateStorePortReservationRequest(binding, config.owner) + _, reserveErr := config.store.CreateOrUpdatePortReservation(reservationCtx, request) + if reserveErr != nil { + if errors.Is(reserveErr, statestore.ErrPortReservationHeld) { + return fmt.Errorf("port %d is already reserved by another DCP process", binding.Port), false + } + return reserveErr, config.mode == portAllocatorModeStateStoreWithFallback + } + return nil, false +} + +func releaseSpecificPortWithStateStore(ctx context.Context, binding ports.Binding, log logr.Logger) (error, bool) { + config, shouldFallback, configErr := getStateStorePortAllocationConfig(log) + if configErr != nil { + return configErr, shouldFallback + } + + if IsEphemeralPort(binding.Port) && !config.allowEphemeralOverlap { + return nil, false + } + + releaseCtx, cancel := context.WithTimeout(ctx, config.allocationTimeout) + defer cancel() + + if lockErr := stateStorePortReservationLock.Lock(releaseCtx); lockErr != nil { + return lockErr, config.mode == portAllocatorModeStateStoreWithFallback + } + defer stateStorePortReservationLock.Unlock() + + request := stateStorePortReservationRequest(binding, config.owner) + releaseErr := config.store.ReleasePort(releaseCtx, request) + if releaseErr != nil { + return releaseErr, config.mode == portAllocatorModeStateStoreWithFallback + } + return nil, false +} + +func stateStorePortReservationRequest(binding ports.Binding, owner process.ProcessTreeItem) statestore.PortReservationRequest { + return statestore.PortReservationRequest{ + Binding: binding, + OwnerProcess: owner, + } +} + +func configuredPortAllocationRanges(allowEphemeralOverlap bool, log logr.Logger) (indexedPortRanges, error) { + configuredRange := strings.TrimSpace(os.Getenv(DCP_PORT_ALLOCATION_RANGE)) + // Default to the upper half of the registered port range: it has a low chance of colliding with + // well-known service ports and stays below the default ephemeral range on almost all supported OSes. + ranges := []portRange{{Start: defaultStateStorePortAllocationRangeStart, End: defaultStateStorePortAllocationRangeEnd}} + if configuredRange != "" { + parsedRanges, parseErr := parsePortAllocationRanges(configuredRange) + if parseErr != nil { + return indexedPortRanges{}, parseErr + } + ranges = parsedRanges + } + + if allowEphemeralOverlap { + return newIndexedPortRanges(ranges), nil + } + + ephemeralStart, ephemeralEnd, matched := GetEphemeralPortRange() + if !matched { + ephemeralStart = DefaultEphemeralPortRangeStart + ephemeralEnd = DefaultEphemeralPortRangeEnd + } + filteredRanges, overlapped := subtractPortRange(ranges, portRange{Start: ephemeralStart, End: ephemeralEnd}) + if overlapped { + reportPortRangeOverlapWarning(log, ephemeralStart, ephemeralEnd) + } + if len(filteredRanges) == 0 { + return indexedPortRanges{}, fmt.Errorf("configured DCP port allocation range overlaps the system ephemeral port range %d-%d; set %s=true to allow overlap", ephemeralStart, ephemeralEnd, DCP_PORT_ALLOCATION_ALLOW_EPHEMERAL_OVERLAP) + } + return newIndexedPortRanges(filteredRanges), nil +} + +func parsePortAllocationRanges(value string) ([]portRange, error) { + parts := strings.Split(value, ",") + ranges := make([]portRange, 0, len(parts)) + for _, part := range parts { + part = strings.TrimSpace(part) + if part == "" { + continue + } + bounds := strings.Split(part, "-") + if len(bounds) != 2 { + return nil, fmt.Errorf("invalid DCP port allocation range '%s'", part) + } + start, startErr := strconv.Atoi(strings.TrimSpace(bounds[0])) + if startErr != nil { + return nil, fmt.Errorf("invalid DCP port allocation range start '%s': %w", bounds[0], startErr) + } + end, endErr := strconv.Atoi(strings.TrimSpace(bounds[1])) + if endErr != nil { + return nil, fmt.Errorf("invalid DCP port allocation range end '%s': %w", bounds[1], endErr) + } + if start > end || !IsValidPort(start) || !IsValidPort(end) { + return nil, fmt.Errorf("invalid DCP port allocation range '%s'", part) + } + ranges = append(ranges, portRange{Start: start, End: end}) + } + if len(ranges) == 0 { + return nil, fmt.Errorf("DCP port allocation range cannot be empty") + } + return ranges, nil +} + +func subtractPortRange(ranges []portRange, excluded portRange) ([]portRange, bool) { + filteredRanges := []portRange{} + overlapped := false + for _, candidate := range ranges { + if candidate.End < excluded.Start || candidate.Start > excluded.End { + filteredRanges = append(filteredRanges, candidate) + continue + } + + overlapped = true + if candidate.Start < excluded.Start { + filteredRanges = append(filteredRanges, portRange{Start: candidate.Start, End: excluded.Start - 1}) + } + if candidate.End > excluded.End { + filteredRanges = append(filteredRanges, portRange{Start: excluded.End + 1, End: candidate.End}) + } + } + return filteredRanges, overlapped +} + +func reportPortRangeOverlapWarning(log logr.Logger, ephemeralStart int, ephemeralEnd int) { + portAllocatorLock.Lock() + defer portAllocatorLock.Unlock() + if portRangeOverlapReported { + return + } + portRangeOverlapReported = true + log.Info("Configured DCP port allocation range overlaps the system ephemeral port range; overlapping ports will be ignored", + "EphemeralRange", fmt.Sprintf("%d-%d", ephemeralStart, ephemeralEnd), + "AllowOverlapEnvVar", DCP_PORT_ALLOCATION_ALLOW_EPHEMERAL_OVERLAP) +} + +type indexedPortRanges struct { + ranges []portRange + total int +} + +func newIndexedPortRanges(ranges []portRange) indexedPortRanges { + if len(ranges) == 0 { + return indexedPortRanges{} + } + + normalizedRanges := append([]portRange{}, ranges...) + sort.Slice(normalizedRanges, func(i int, j int) bool { + if normalizedRanges[i].Start == normalizedRanges[j].Start { + return normalizedRanges[i].End < normalizedRanges[j].End + } + return normalizedRanges[i].Start < normalizedRanges[j].Start + }) + + mergedRanges := make([]portRange, 0, len(normalizedRanges)) + for _, candidateRange := range normalizedRanges { + if len(mergedRanges) == 0 { + mergedRanges = append(mergedRanges, candidateRange) + continue + } + + lastRangeIndex := len(mergedRanges) - 1 + if candidateRange.Start <= mergedRanges[lastRangeIndex].End+1 { + if candidateRange.End > mergedRanges[lastRangeIndex].End { + mergedRanges[lastRangeIndex].End = candidateRange.End + } + continue + } + mergedRanges = append(mergedRanges, candidateRange) + } + + total := 0 + for _, candidateRange := range mergedRanges { + total += candidateRange.End - candidateRange.Start + 1 + } + + return indexedPortRanges{ + ranges: mergedRanges, + total: total, + } +} + +func (ranges indexedPortRanges) Len() int { + return ranges.total +} + +func (ranges indexedPortRanges) PortAt(index int) (int32, bool) { + if index < 0 || index >= ranges.total { + return 0, false + } + + candidateIndex := index + for _, candidateRange := range ranges.ranges { + rangeLength := candidateRange.End - candidateRange.Start + 1 + if candidateIndex < rangeLength { + return int32(candidateRange.Start + candidateIndex), true + } + candidateIndex -= rangeLength + } + + return 0, false +} + +type stateStorePortAllocationCandidateIterator struct { + ranges indexedPortRanges + total int + offset int + step int + next int +} + +// newStateStorePortAllocationCandidateIterator creates a deterministic random walk over the +// configured ranges. The walk is lazy so large ranges do not need to be materialized before the +// allocator finds an available port. +func newStateStorePortAllocationCandidateIterator(ranges indexedPortRanges, protocol string, address string) *stateStorePortAllocationCandidateIterator { + total := ranges.Len() + if total <= 0 { + return &stateStorePortAllocationCandidateIterator{} + } + + hash := fnv.New64a() + _, _ = hash.Write([]byte(programInstanceID)) + _, _ = hash.Write([]byte(protocol)) + _, _ = hash.Write([]byte(address)) + hashValue := hash.Sum64() + offset := int(hashValue % uint64(total)) + step := stateStorePortAllocationCandidateStep(total, hashValue/uint64(total)) + + return &stateStorePortAllocationCandidateIterator{ + ranges: ranges, + total: total, + offset: offset, + step: step, + } +} + +// Next returns the next candidate port from the virtual flattened port range. +func (iterator *stateStorePortAllocationCandidateIterator) Next() (int32, bool) { + if iterator.next >= iterator.total { + return 0, false + } + + candidateIndex := (iterator.offset + iterator.next*iterator.step) % iterator.total + iterator.next++ + return iterator.ranges.PortAt(candidateIndex) +} + +// stateStorePortAllocationCandidateStep returns a pseudo-random step that is coprime with the +// number of candidates. A coprime step guarantees the modular walk visits every candidate exactly +// once instead of cycling through a subset of the range. +func stateStorePortAllocationCandidateStep(total int, hashValue uint64) int { + if total <= 1 { + return 1 + } + + step := int(hashValue%uint64(total-1)) + 1 + for gcd(step, total) != 1 { + step++ + if step >= total { + step = 1 + } + } + return step +} + +// gcd computes the greatest common divisor using Euclid's algorithm. +func gcd(a int, b int) int { + for b != 0 { + a, b = b, a%b + } + if a < 0 { + return -a + } + return a +} diff --git a/internal/proxy/proxy_test.go b/internal/proxy/proxy_test.go index e8ff0057..18e4392c 100644 --- a/internal/proxy/proxy_test.go +++ b/internal/proxy/proxy_test.go @@ -159,7 +159,7 @@ func TestTCPClientCanSendDataBeforeEndpointsExist(t *testing.T) { } // Configure the proxy with valid endpoint - port, err := networking.GetFreePort(apiv1.TCP, networking.IPv4LocalhostDefaultAddress, log) + port, err := networking.GetFreePort(context.Background(), apiv1.TCP, networking.IPv4LocalhostDefaultAddress, log) require.NoError(t, err) config := ProxyConfig{ Endpoints: []Endpoint{ @@ -693,7 +693,7 @@ func TestTCPProxyContinuousStream(t *testing.T) { t.Parallel() - serverPort, portErr := networking.GetFreePort(apiv1.TCP, networking.IPv4LocalhostDefaultAddress, log) + serverPort, portErr := networking.GetFreePort(context.Background(), apiv1.TCP, networking.IPv4LocalhostDefaultAddress, log) require.NoError(t, portErr, "Failed to get free port for server") // Set up a proxy diff --git a/internal/statestore/migrations/000002_port_allocations.up.sql b/internal/statestore/migrations/000002_port_allocations.up.sql new file mode 100644 index 00000000..cc884b4c --- /dev/null +++ b/internal/statestore/migrations/000002_port_allocations.up.sql @@ -0,0 +1,11 @@ +CREATE TABLE IF NOT EXISTS port_allocations ( + protocol TEXT NOT NULL, + ip BLOB NOT NULL CHECK(length(ip) IN (4, 16)), + port INTEGER NOT NULL, + owner_pid INTEGER NOT NULL, + owner_identity_time TEXT NOT NULL, + updated_at_unix_nano INTEGER NOT NULL, + PRIMARY KEY(protocol, ip, port) +); + +CREATE INDEX IF NOT EXISTS idx_port_allocations_owner ON port_allocations(owner_pid, owner_identity_time); diff --git a/internal/statestore/ports.go b/internal/statestore/ports.go new file mode 100644 index 00000000..b2331df0 --- /dev/null +++ b/internal/statestore/ports.go @@ -0,0 +1,519 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package statestore + +import ( + "bytes" + "context" + "database/sql" + "errors" + "fmt" + "net/netip" + "strings" + "time" + + "github.com/microsoft/dcp/pkg/ports" + "github.com/microsoft/dcp/pkg/process" +) + +const ( + ipv4AddressBytes = 4 + ipv6AddressBytes = 16 +) + +var ErrPortReservationHeld = errors.New("port reservation is held by another owner") + +// PortReservation describes a state-store claim on a protocol/IP/port tuple by a DCP owner process. +type PortReservation struct { + ports.Binding + OwnerProcess process.ProcessTreeItem + UpdatedAt time.Time +} + +// PortReservationRequest identifies the protocol/IP/port tuple to reserve or release and the DCP +// owner process responsible for that claim. +type PortReservationRequest struct { + ports.Binding + OwnerProcess process.ProcessTreeItem +} + +// PortReservationHeldError reports the conflicting active reservation when a requested +// protocol/IP/port tuple is already held by another DCP owner. +type PortReservationHeldError struct { + Reservation PortReservation +} + +func (e *PortReservationHeldError) Error() string { + return fmt.Sprintf("%s: %s/%s/%d", ErrPortReservationHeld, e.Reservation.Protocol, e.Reservation.IP, e.Reservation.Port) +} + +func (e *PortReservationHeldError) Unwrap() error { + return ErrPortReservationHeld +} + +// HeldPortReservation extracts the active reservation that blocked a reserve attempt. +func HeldPortReservation(err error) (*PortReservation, bool) { + var heldErr *PortReservationHeldError + if !errors.As(err, &heldErr) { + return nil, false + } + + reservation := heldErr.Reservation + return &reservation, true +} + +// CreatePortReservation creates a reservation for the requested protocol/IP/port tuple. +// It fails with ErrPortReservationHeld when any active owner already reserves the same tuple or a +// same-family wildcard tuple for that port. +func (s *Store) CreatePortReservation(ctx context.Context, request PortReservationRequest) (*PortReservation, error) { + return s.reservePort(ctx, request, false) +} + +// CreateOrUpdatePortReservation creates a reservation for the requested protocol/IP/port tuple. +// If the same owner already holds the exact tuple, it refreshes that reservation instead. It still +// rejects wildcard/specific conflicts so one owner cannot hold overlapping claims for the same port. +func (s *Store) CreateOrUpdatePortReservation(ctx context.Context, request PortReservationRequest) (*PortReservation, error) { + return s.reservePort(ctx, request, true) +} + +// ReleasePort removes this owner process' exact protocol/IP/port reservation. +// Reservations held by other owners, or overlapping wildcard/specific reservations, are left intact. +func (s *Store) ReleasePort(ctx context.Context, request PortReservationRequest) error { + normalizedRequest, requestErr := normalizePortReservationRequest(request) + if requestErr != nil { + return requestErr + } + + return s.withImmediateTx(ctx, func(conn *sql.Conn) error { + return deleteExactPortReservation(ctx, conn, normalizedRequest) + }) +} + +// DeleteInactivePortReservations deletes reservations whose owner process identity is no longer +// active, allowing ports abandoned by exited DCP instances to be reused. +func (s *Store) DeleteInactivePortReservations(ctx context.Context) error { + candidates, candidatesErr := s.inactivePortReservationCandidates(ctx) + if candidatesErr != nil { + return candidatesErr + } + if len(candidates) == 0 { + return nil + } + + return s.withImmediateTx(ctx, func(conn *sql.Conn) error { + for _, candidate := range candidates { + _, execErr := conn.ExecContext( + ctx, + `DELETE FROM port_allocations + WHERE protocol = ? AND ip = ? AND port = ? + AND owner_pid = ? AND owner_identity_time = ? + AND updated_at_unix_nano = ?`, + candidate.Protocol, + portReservationIPBytes(candidate.IP), + candidate.Port, + candidate.OwnerProcess.Pid, + timeString(candidate.OwnerProcess.IdentityTime), + unixNano(candidate.UpdatedAt), + ) + if execErr != nil { + return fmt.Errorf("could not delete inactive port reservation '%s/%s/%d': %w", candidate.Protocol, candidate.IP, candidate.Port, execErr) + } + } + return nil + }) +} + +func (s *Store) reservePort(ctx context.Context, request PortReservationRequest, updateIfExisting bool) (*PortReservation, error) { + normalizedRequest, requestErr := normalizePortReservationRequest(request) + if requestErr != nil { + return nil, requestErr + } + + now := time.Now().UTC() + var reservation *PortReservation + txErr := s.withImmediateTx(ctx, func(conn *sql.Conn) error { + requestKey := normalizedPortReservationKey{ + protocol: normalizedRequest.Protocol, + ip: normalizedRequest.IP, + ipBytes: portReservationIPBytes(normalizedRequest.IP), + port: normalizedRequest.Port, + } + + conflictingReservations, conflictsErr := getConflictingPortReservations(ctx, conn, requestKey) + if conflictsErr != nil { + return conflictsErr + } + if len(conflictingReservations) == 0 { + insertErr := insertPortReservation(ctx, conn, normalizedRequest, now) + if insertErr != nil { + return insertErr + } + var readErr error + reservation, _, readErr = getExactPortReservation(ctx, conn, normalizedRequest.Protocol, normalizedRequest.IP, normalizedRequest.Port) + return readErr + } + + inactiveReservations := make([]*PortReservation, 0, len(conflictingReservations)) + shouldUpdateExactReservation := false + for _, conflictingReservation := range conflictingReservations { + exactIP := conflictingReservation.IP == normalizedRequest.IP + sameOwner := conflictingReservation.OwnerProcess.Pid == normalizedRequest.OwnerProcess.Pid && + conflictingReservation.OwnerProcess.IdentityTime.Equal(normalizedRequest.OwnerProcess.IdentityTime) + if updateIfExisting && sameOwner { + if exactIP { + shouldUpdateExactReservation = true + continue + } + } + active := resourceLeaseOwnerIsActive(conflictingReservation.OwnerProcess) + if !active { + if exactIP { + shouldUpdateExactReservation = true + } else { + inactiveReservations = append(inactiveReservations, conflictingReservation) + } + continue + } + + return &PortReservationHeldError{Reservation: *conflictingReservation} + } + for _, conflictingReservation := range inactiveReservations { + deleteErr := deletePortReservation(ctx, conn, conflictingReservation) + if deleteErr != nil { + return deleteErr + } + } + if shouldUpdateExactReservation { + updateErr := updatePortReservation(ctx, conn, normalizedRequest, now) + if updateErr != nil { + return updateErr + } + } else { + insertErr := insertPortReservation(ctx, conn, normalizedRequest, now) + if insertErr != nil { + return insertErr + } + } + var readErr error + reservation, _, readErr = getExactPortReservation(ctx, conn, normalizedRequest.Protocol, normalizedRequest.IP, normalizedRequest.Port) + return readErr + }) + if txErr != nil { + return nil, txErr + } + + return reservation, nil +} + +func normalizePortReservationRequest(request PortReservationRequest) (PortReservationRequest, error) { + key, keyErr := normalizePortReservationKey(request.Binding) + if keyErr != nil { + return PortReservationRequest{}, keyErr + } + normalizedOwner, ownerErr := normalizeResourceLeaseOwner(request.OwnerProcess) + if ownerErr != nil { + return PortReservationRequest{}, fmt.Errorf("%w: %w", ErrInvalidArgument, ownerErr) + } + + return PortReservationRequest{ + Binding: ports.Binding{ + Protocol: key.protocol, + IP: key.ip, + Port: key.port, + }, + OwnerProcess: normalizedOwner, + }, nil +} + +type normalizedPortReservationKey struct { + protocol string + ip netip.Addr + ipBytes []byte + port int32 +} + +func normalizePortReservationKey(binding ports.Binding) (normalizedPortReservationKey, error) { + protocol := strings.ToUpper(strings.TrimSpace(binding.Protocol)) + if protocol == "" { + return normalizedPortReservationKey{}, fmt.Errorf("%w: port reservation protocol cannot be empty", ErrInvalidArgument) + } + if !binding.IP.IsValid() { + return normalizedPortReservationKey{}, fmt.Errorf("%w: port reservation ip cannot be empty", ErrInvalidArgument) + } + if !ports.IsValidPort(int(binding.Port)) { + return normalizedPortReservationKey{}, fmt.Errorf("%w: port reservation port must be between 1 and 65535", ErrInvalidArgument) + } + addr := binding.IP.Unmap() + return normalizedPortReservationKey{ + protocol: protocol, + ip: addr, + ipBytes: portReservationIPBytes(addr), + port: binding.Port, + }, nil +} + +func portReservationIPBytes(addr netip.Addr) []byte { + if addr.Is4() { + bytes := addr.As4() + return append([]byte(nil), bytes[:]...) + } + bytes := addr.As16() + return append([]byte(nil), bytes[:]...) +} + +func portReservationAddrFromBytes(ipBytes []byte) (netip.Addr, error) { + switch len(ipBytes) { + case ipv4AddressBytes: + var bytes [ipv4AddressBytes]byte + copy(bytes[:], ipBytes) + return netip.AddrFrom4(bytes), nil + case ipv6AddressBytes: + var bytes [ipv6AddressBytes]byte + copy(bytes[:], ipBytes) + return netip.AddrFrom16(bytes), nil + default: + return netip.Addr{}, fmt.Errorf("port reservation ip has invalid byte length %d", len(ipBytes)) + } +} + +type portReservationScanner interface { + Scan(dest ...any) error +} + +type portReservationQuerier interface { + QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) + QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row +} + +func getExactPortReservation(ctx context.Context, querier portReservationQuerier, protocol string, ip netip.Addr, port int32) (*PortReservation, bool, error) { + key, keyErr := normalizePortReservationKey(ports.Binding{Protocol: protocol, IP: ip, Port: port}) + if keyErr != nil { + return nil, false, keyErr + } + row := querier.QueryRowContext( + ctx, + `SELECT protocol, ip, port, + owner_pid, owner_identity_time, updated_at_unix_nano + FROM port_allocations + WHERE protocol = ? AND ip = ? AND port = ?`, + key.protocol, + key.ipBytes, + key.port, + ) + + reservation, scanErr := scanPortReservation(row) + if errors.Is(scanErr, sql.ErrNoRows) { + return nil, false, nil + } + if scanErr != nil { + return nil, false, fmt.Errorf("could not read port reservation '%s/%s/%d': %w", key.protocol, key.ip, key.port, scanErr) + } + return reservation, true, nil +} + +func getConflictingPortReservations(ctx context.Context, querier portReservationQuerier, key normalizedPortReservationKey) ([]*PortReservation, error) { + rows, queryErr := querier.QueryContext( + ctx, + `SELECT protocol, ip, port, + owner_pid, owner_identity_time, updated_at_unix_nano + FROM port_allocations + WHERE protocol = ? AND port = ?`, + key.protocol, + key.port, + ) + if queryErr != nil { + return nil, fmt.Errorf("could not list port reservations for '%s/%s/%d': %w", key.protocol, key.ip, key.port, queryErr) + } + defer func() { + _ = rows.Close() + }() + + reservations := []*PortReservation{} + for rows.Next() { + reservation, scanErr := scanPortReservation(rows) + if scanErr != nil { + return nil, fmt.Errorf("could not read port reservation for '%s/%s/%d': %w", key.protocol, key.ip, key.port, scanErr) + } + if portReservationIPsConflict(key.ipBytes, portReservationIPBytes(reservation.IP)) { + reservations = append(reservations, reservation) + } + } + if rowsErr := rows.Err(); rowsErr != nil { + return nil, fmt.Errorf("could not list port reservations for '%s/%s/%d': %w", key.protocol, key.ip, key.port, rowsErr) + } + + return reservations, nil +} + +func portReservationIPsConflict(requestedIP []byte, reservedIP []byte) bool { + if bytes.Equal(requestedIP, reservedIP) { + return true + } + if len(requestedIP) != len(reservedIP) { + return false + } + + requestedAddr, requestedErr := portReservationAddrFromBytes(requestedIP) + if requestedErr != nil { + return false + } + reservedAddr, reservedErr := portReservationAddrFromBytes(reservedIP) + if reservedErr != nil { + return false + } + + return requestedAddr.IsUnspecified() || reservedAddr.IsUnspecified() +} + +func scanPortReservation(row portReservationScanner) (*PortReservation, error) { + var reservation PortReservation + var ipBytes []byte + var port int64 + var ownerPID int64 + var ownerIdentityTime string + var updatedAtUnixNano int64 + + scanErr := row.Scan( + &reservation.Protocol, + &ipBytes, + &port, + &ownerPID, + &ownerIdentityTime, + &updatedAtUnixNano, + ) + if scanErr != nil { + return nil, scanErr + } + if !ports.IsValidPort(int(port)) { + return nil, fmt.Errorf("port reservation port is out of range: %d", port) + } + ip, ipErr := portReservationAddrFromBytes(ipBytes) + if ipErr != nil { + return nil, ipErr + } + reservation.IP = ip + reservation.Port = int32(port) + + ownerProcess, ownerErr := resourceLeaseOwnerFromDB(ownerPID, ownerIdentityTime) + if ownerErr != nil { + return nil, fmt.Errorf("could not read port reservation owner: %w", ownerErr) + } + reservation.OwnerProcess = ownerProcess + reservation.UpdatedAt = timeFromUnixNano(updatedAtUnixNano) + return &reservation, nil +} + +func insertPortReservation(ctx context.Context, conn *sql.Conn, request PortReservationRequest, now time.Time) error { + _, insertErr := conn.ExecContext( + ctx, + `INSERT INTO port_allocations( + protocol, ip, port, + owner_pid, owner_identity_time, updated_at_unix_nano + ) + VALUES(?, ?, ?, ?, ?, ?)`, + request.Protocol, + portReservationIPBytes(request.IP), + request.Port, + request.OwnerProcess.Pid, + timeString(request.OwnerProcess.IdentityTime), + unixNano(now), + ) + if insertErr != nil { + return fmt.Errorf("could not insert port reservation '%s/%s/%d': %w", request.Protocol, request.IP, request.Port, insertErr) + } + return nil +} + +func updatePortReservation(ctx context.Context, conn *sql.Conn, request PortReservationRequest, now time.Time) error { + _, updateErr := conn.ExecContext( + ctx, + `UPDATE port_allocations + SET owner_pid = ?, owner_identity_time = ?, updated_at_unix_nano = ? + WHERE protocol = ? AND ip = ? AND port = ?`, + request.OwnerProcess.Pid, + timeString(request.OwnerProcess.IdentityTime), + unixNano(now), + request.Protocol, + portReservationIPBytes(request.IP), + request.Port, + ) + if updateErr != nil { + return fmt.Errorf("could not update port reservation '%s/%s/%d': %w", request.Protocol, request.IP, request.Port, updateErr) + } + return nil +} + +func deletePortReservation(ctx context.Context, conn *sql.Conn, reservation *PortReservation) error { + _, deleteErr := conn.ExecContext( + ctx, + `DELETE FROM port_allocations + WHERE protocol = ? AND ip = ? AND port = ? + AND owner_pid = ? AND owner_identity_time = ?`, + reservation.Protocol, + portReservationIPBytes(reservation.IP), + reservation.Port, + reservation.OwnerProcess.Pid, + timeString(reservation.OwnerProcess.IdentityTime), + ) + if deleteErr != nil { + return fmt.Errorf("could not delete port reservation '%s/%s/%d': %w", reservation.Protocol, reservation.IP, reservation.Port, deleteErr) + } + return nil +} + +func deleteExactPortReservation(ctx context.Context, conn *sql.Conn, request PortReservationRequest) error { + _, deleteErr := conn.ExecContext( + ctx, + `DELETE FROM port_allocations + WHERE protocol = ? AND ip = ? AND port = ? + AND owner_pid = ? AND owner_identity_time = ?`, + request.Protocol, + portReservationIPBytes(request.IP), + request.Port, + request.OwnerProcess.Pid, + timeString(request.OwnerProcess.IdentityTime), + ) + if deleteErr != nil { + return fmt.Errorf("could not delete port reservation '%s/%s/%d': %w", request.Protocol, request.IP, request.Port, deleteErr) + } + return nil +} + +func (s *Store) inactivePortReservationCandidates(ctx context.Context) ([]PortReservation, error) { + db, dbErr := s.requireDB() + if dbErr != nil { + return nil, dbErr + } + + rows, queryErr := db.QueryContext( + ctx, + `SELECT protocol, ip, port, + owner_pid, owner_identity_time, updated_at_unix_nano + FROM port_allocations`, + ) + if queryErr != nil { + return nil, fmt.Errorf("could not list port reservations for cleanup: %w", queryErr) + } + defer func() { + _ = rows.Close() + }() + + candidates := []PortReservation{} + for rows.Next() { + reservation, scanErr := scanPortReservation(rows) + if scanErr != nil { + return nil, fmt.Errorf("could not read port reservation for cleanup: %w", scanErr) + } + if !resourceLeaseOwnerIsActive(reservation.OwnerProcess) { + candidates = append(candidates, *reservation) + } + } + if rowsErr := rows.Err(); rowsErr != nil { + return nil, fmt.Errorf("could not list port reservations for cleanup: %w", rowsErr) + } + + return candidates, nil +} diff --git a/internal/statestore/schema.go b/internal/statestore/schema.go index 777f1fba..0b9c7630 100644 --- a/internal/statestore/schema.go +++ b/internal/statestore/schema.go @@ -21,7 +21,7 @@ import ( ) const ( - currentSchemaVersion = 1 + currentSchemaVersion = 2 migrationTableName = "schema_migrations" migrationLockFileSuffix = ".migrate.lock" diff --git a/internal/statestore/store_test.go b/internal/statestore/store_test.go index 38228de7..93266820 100644 --- a/internal/statestore/store_test.go +++ b/internal/statestore/store_test.go @@ -10,6 +10,7 @@ import ( "database/sql" "errors" "io/fs" + "net/netip" "os" "path/filepath" "runtime" @@ -20,12 +21,21 @@ import ( usvc_io "github.com/microsoft/dcp/pkg/io" "github.com/microsoft/dcp/pkg/osutil" + "github.com/microsoft/dcp/pkg/ports" "github.com/microsoft/dcp/pkg/process" "github.com/microsoft/dcp/pkg/testutil" ) const stateStoreTestTimeout = 10 * time.Second +func testPortBinding(protocol string, ip string, port int32) ports.Binding { + return ports.Binding{ + Protocol: protocol, + IP: netip.MustParseAddr(ip), + Port: port, + } +} + func openRawSQLiteDB(t *testing.T, ctx context.Context, path string) *sql.DB { t.Helper() @@ -545,6 +555,349 @@ func TestDeleteInactiveResourceLeasesUsesOwnerProcessIdentity(t *testing.T) { require.NoError(t, invalidOwnerReacquireErr) } +func TestCreatePortReservationBlocksOtherStore(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + owner1, owner1Err := testResourceLeaseOwner(t, 0) + require.NoError(t, owner1Err) + owner2, owner2Err := testResourceLeaseOwner(t, time.Second) + require.NoError(t, owner2Err) + + _, reserveErr := store1.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26001), + OwnerProcess: owner1, + }) + require.NoError(t, reserveErr) + + _, blockedErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26001), + OwnerProcess: owner2, + }) + + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) +} + +func TestCreateOrUpdatePortReservationReusesSameOwner(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store := openTestStore(t, ctx, storePath) + owner, ownerErr := testResourceLeaseOwner(t, 0) + require.NoError(t, ownerErr) + + first, firstErr := store.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26002), + OwnerProcess: owner, + }) + require.NoError(t, firstErr) + second, secondErr := store.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26002), + OwnerProcess: owner, + }) + + require.NoError(t, secondErr) + require.Equal(t, first.Protocol, second.Protocol) + require.Equal(t, first.IP, second.IP) + require.Equal(t, first.Port, second.Port) +} + +func TestReleasePortAllowsOtherOwnerToReserve(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + owner1, owner1Err := testResourceLeaseOwner(t, 0) + require.NoError(t, owner1Err) + owner2, owner2Err := testResourceLeaseOwner(t, time.Second) + require.NoError(t, owner2Err) + + _, reserveErr := store1.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26005), + OwnerProcess: owner1, + }) + require.NoError(t, reserveErr) + _, blockedErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26005), + OwnerProcess: owner2, + }) + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) + + releaseErr := store1.ReleasePort(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26005), + OwnerProcess: owner1, + }) + require.NoError(t, releaseErr) + + _, reserveAfterReleaseErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26005), + OwnerProcess: owner2, + }) + require.NoError(t, reserveAfterReleaseErr) +} + +func TestReleasePortDoesNotReleaseOtherOwnerReservation(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + owner1, owner1Err := testResourceLeaseOwner(t, 0) + require.NoError(t, owner1Err) + owner2, owner2Err := testResourceLeaseOwner(t, time.Second) + require.NoError(t, owner2Err) + + _, reserveErr := store1.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26006), + OwnerProcess: owner1, + }) + require.NoError(t, reserveErr) + releaseErr := store2.ReleasePort(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26006), + OwnerProcess: owner2, + }) + require.NoError(t, releaseErr) + + _, blockedErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26006), + OwnerProcess: owner2, + }) + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) +} + +func TestPortReservationAllIPv4InterfacesBlocksSpecificAddress(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + owner1, owner1Err := testResourceLeaseOwner(t, 0) + require.NoError(t, owner1Err) + owner2, owner2Err := testResourceLeaseOwner(t, time.Second) + require.NoError(t, owner2Err) + + _, reserveErr := store1.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "0.0.0.0", 26008), + OwnerProcess: owner1, + }) + require.NoError(t, reserveErr) + + _, blockedErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26008), + OwnerProcess: owner2, + }) + + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) +} + +func TestPortReservationSpecificAddressBlocksAllIPv4Interfaces(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + owner1, owner1Err := testResourceLeaseOwner(t, 0) + require.NoError(t, owner1Err) + owner2, owner2Err := testResourceLeaseOwner(t, time.Second) + require.NoError(t, owner2Err) + + _, reserveErr := store1.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26009), + OwnerProcess: owner1, + }) + require.NoError(t, reserveErr) + + _, blockedErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "0.0.0.0", 26009), + OwnerProcess: owner2, + }) + + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) +} + +func TestCreateOrUpdatePortReservationSameOwnerSpecificAddressBlocksAllIPv4Interfaces(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store := openTestStore(t, ctx, storePath) + owner, ownerErr := testResourceLeaseOwner(t, 0) + require.NoError(t, ownerErr) + + _, reserveErr := store.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26012), + OwnerProcess: owner, + }) + require.NoError(t, reserveErr) + + _, blockedErr := store.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "0.0.0.0", 26012), + OwnerProcess: owner, + }) + + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) +} + +func TestCreateOrUpdatePortReservationSameOwnerAllIPv4InterfacesBlocksSpecificAddress(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store := openTestStore(t, ctx, storePath) + owner, ownerErr := testResourceLeaseOwner(t, 0) + require.NoError(t, ownerErr) + + _, reserveErr := store.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "0.0.0.0", 26013), + OwnerProcess: owner, + }) + require.NoError(t, reserveErr) + + _, blockedErr := store.CreateOrUpdatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26013), + OwnerProcess: owner, + }) + + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) +} + +func TestPortReservationAllIPv6InterfacesBlocksSpecificAddress(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + owner1, owner1Err := testResourceLeaseOwner(t, 0) + require.NoError(t, owner1Err) + owner2, owner2Err := testResourceLeaseOwner(t, time.Second) + require.NoError(t, owner2Err) + + _, reserveErr := store1.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "::", 26010), + OwnerProcess: owner1, + }) + require.NoError(t, reserveErr) + + _, blockedErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "::1", 26010), + OwnerProcess: owner2, + }) + + require.ErrorIs(t, blockedErr, ErrPortReservationHeld) +} + +func TestPortReservationIPv4WildcardDoesNotBlockIPv6(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + owner1, owner1Err := testResourceLeaseOwner(t, 0) + require.NoError(t, owner1Err) + owner2, owner2Err := testResourceLeaseOwner(t, time.Second) + require.NoError(t, owner2Err) + + _, reserveErr := store1.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "0.0.0.0", 26011), + OwnerProcess: owner1, + }) + require.NoError(t, reserveErr) + + reservation, reserveIPv6Err := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "::1", 26011), + OwnerProcess: owner2, + }) + + require.NoError(t, reserveIPv6Err) + require.Equal(t, netip.MustParseAddr("::1"), reservation.IP) +} + +func TestPortReservationStoresIPAsBlob(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store := openTestStore(t, ctx, storePath) + owner, ownerErr := testResourceLeaseOwner(t, 0) + require.NoError(t, ownerErr) + + reservation, reserveErr := store.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "::1", 26007), + OwnerProcess: owner, + }) + require.NoError(t, reserveErr) + + require.Equal(t, netip.MustParseAddr("::1"), reservation.IP) + row := store.db.QueryRowContext(ctx, `SELECT typeof(ip), length(ip) FROM port_allocations WHERE protocol = ? AND port = ?`, "TCP", 26007) + var addressType string + var addressLength int + require.NoError(t, row.Scan(&addressType, &addressLength)) + require.Equal(t, "blob", addressType) + require.Equal(t, 16, addressLength) +} + +func TestDeleteInactivePortReservationsUsesOwnerProcessIdentity(t *testing.T) { + t.Parallel() + + ctx, cancel := testutil.GetTestContext(t, stateStoreTestTimeout) + defer cancel() + storePath := filepath.Join(t.TempDir(), "state.sqlite3") + store1 := openTestStore(t, ctx, storePath) + store2 := openTestStore(t, ctx, storePath) + activeOwner, activeOwnerErr := testResourceLeaseOwner(t, 0) + require.NoError(t, activeOwnerErr) + staleOwner, staleOwnerErr := testResourceLeaseOwner(t, time.Second) + require.NoError(t, staleOwnerErr) + otherOwner, otherOwnerErr := testResourceLeaseOwner(t, 2*time.Second) + require.NoError(t, otherOwnerErr) + + _, activeReserveErr := store1.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26003), + OwnerProcess: activeOwner, + }) + require.NoError(t, activeReserveErr) + _, staleReserveErr := store1.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26004), + OwnerProcess: staleOwner, + }) + require.NoError(t, staleReserveErr) + + require.NoError(t, store1.DeleteInactivePortReservations(ctx)) + + _, activeBlockedErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26003), + OwnerProcess: otherOwner, + }) + require.ErrorIs(t, activeBlockedErr, ErrPortReservationHeld) + _, staleReacquireErr := store2.CreatePortReservation(ctx, PortReservationRequest{ + Binding: testPortBinding("tcp", "127.0.0.1", 26004), + OwnerProcess: otherOwner, + }) + require.NoError(t, staleReacquireErr) +} + func TestPersistentProcessRecordRoundTrip(t *testing.T) { t.Parallel() diff --git a/internal/templating/value_templates.go b/internal/templating/value_templates.go index a4c4e19c..3f92eebd 100644 --- a/internal/templating/value_templates.go +++ b/internal/templating/value_templates.go @@ -233,7 +233,7 @@ func portForServingFromExecutable( return 0, errServiceDoesNotExist{Service: serviceName, ContextObject: exe.NamespacedName()} } - port, err := networking.GetFreePort(svc.Spec.Protocol, sp.Address, log) + port, err := networking.GetFreePort(ctx, svc.Spec.Protocol, sp.Address, log) if err != nil { return 0, fmt.Errorf( "could not allocate a port for service '%s' with desired address '%s': %w", diff --git a/pkg/kubeconfig/flag.go b/pkg/kubeconfig/flag.go index 7f3a55d0..c0637c1c 100644 --- a/pkg/kubeconfig/flag.go +++ b/pkg/kubeconfig/flag.go @@ -155,9 +155,8 @@ func RequireKubeconfigFlagValue(flags *pflag.FlagSet) (string, error) { // The kubeconfig flag value, if empty upon invocation, will be set to preferred path of the kubeconfig file. // Does NOT create the kubeconfig file itself (see Kubeconfig.Save() for that). // -// traceCtx is used to propagate trace context for sub-instrumentation; cancellation is not -// observed by this function (the work is short and synchronous). -func EnsureKubeconfigData(traceCtx context.Context, flags *pflag.FlagSet, log logr.Logger) (*Kubeconfig, error) { +// ctx is used to propagate trace context for sub-instrumentation and cancellation for port allocation. +func EnsureKubeconfigData(ctx context.Context, flags *pflag.FlagSet, log logr.Logger) (*Kubeconfig, error) { f := flags.Lookup(ctrl_config.KubeconfigFlagName) if f == nil { return nil, fmt.Errorf("unable to find kubeconfig flag. Make sure you call EnsureKubeconfigFlag() before calling this function.") @@ -177,7 +176,7 @@ func EnsureKubeconfigData(traceCtx context.Context, flags *pflag.FlagSet, log lo } if hasTLSCertFile { - lookup, lookupErr := traced(traceCtx, spanTlsCertLookup, func() (certLookup, error) { + lookup, lookupErr := traced(ctx, spanTlsCertLookup, func() (certLookup, error) { cd, addr, err := security.LoadCertificateFiles(tlsCertFile, tlsKeyFile, tlsCAFile, tlsCertThumbprint) return certLookup{data: cd, address: addr}, err }) @@ -186,7 +185,7 @@ func EnsureKubeconfigData(traceCtx context.Context, flags *pflag.FlagSet, log lo } storeCertData, serverAddress = lookup.data, lookup.address } else if tlsCertThumbprint != "" { - lookup, lookupErr := traced(traceCtx, spanTlsCertLookup, func() (certLookup, error) { + lookup, lookupErr := traced(ctx, spanTlsCertLookup, func() (certLookup, error) { cd, addr, err := security.LookupCertificate(tlsCertThumbprint) return certLookup{data: cd, address: addr}, err }) @@ -207,7 +206,7 @@ func EnsureKubeconfigData(traceCtx context.Context, flags *pflag.FlagSet, log lo if tlsCAFile != "" { return nil, fmt.Errorf("--%s requires --%s/--%s or --%s to also be specified", TLSCAFileFlagName, TLSCertFileFlagName, TLSKeyFileFlagName, TLSCertThumbprintFlagName) } - preferredAddress, preferredErr := traced(traceCtx, spanPreferredHostIps, func() (string, error) { + preferredAddress, preferredErr := traced(ctx, spanPreferredHostIps, func() (string, error) { ips, err := networking.GetPreferredHostIps(networking.Localhost) if err != nil { return "", err @@ -236,7 +235,7 @@ func EnsureKubeconfigData(traceCtx context.Context, flags *pflag.FlagSet, log lo generateEphemeral := storeCertData == nil - k, kErr := getKubeconfig(traceCtx, kubeconfigPath, port, generateEphemeral, generateToken, storeCertData, serverAddress, log) + k, kErr := getKubeconfig(ctx, kubeconfigPath, port, generateEphemeral, generateToken, storeCertData, serverAddress, log) if kErr != nil { return nil, fmt.Errorf("unable to obtain Kubeconfig data: %w", kErr) } diff --git a/pkg/kubeconfig/kubeconfig.go b/pkg/kubeconfig/kubeconfig.go index 6d2488c3..b86a6313 100644 --- a/pkg/kubeconfig/kubeconfig.go +++ b/pkg/kubeconfig/kubeconfig.go @@ -188,7 +188,7 @@ func getKubeconfig(ctx context.Context, kubeconfigPath string, port int32, gener func createKubeconfig(ctx context.Context, port int32, generateEphemeral bool, generateToken bool, storeCertData *security.ServerCertificateData, serverAddress string, log logr.Logger) (*clientcmd_api.Config, *security.ServerCertificateData, error) { if port == 0 { newPort, newPortErr := traced(ctx, spanGetFreePort, func() (int32, error) { - return networking.GetFreePort(apiv1.TCP, serverAddress, log) + return networking.GetFreePort(ctx, apiv1.TCP, serverAddress, log) }) if newPortErr != nil { return nil, nil, newPortErr diff --git a/pkg/ports/ports.go b/pkg/ports/ports.go new file mode 100644 index 00000000..ecdaed35 --- /dev/null +++ b/pkg/ports/ports.go @@ -0,0 +1,23 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package ports + +import "net/netip" + +// Binding identifies a protocol/IP/port tuple. +type Binding struct { + Protocol string + IP netip.Addr + Port int32 +} + +func IsValidPort(port int) bool { + return port >= 1 && port <= 65535 +} + +func IsBindablePort(port int) bool { + return port == 0 || IsValidPort(port) +} diff --git a/pkg/ports/ports_test.go b/pkg/ports/ports_test.go new file mode 100644 index 00000000..104838cc --- /dev/null +++ b/pkg/ports/ports_test.go @@ -0,0 +1,30 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) Microsoft Corporation. All rights reserved. + * Licensed under the MIT License. See LICENSE in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package ports + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestIsValidPort(t *testing.T) { + t.Parallel() + + require.False(t, IsValidPort(0)) + require.True(t, IsValidPort(1)) + require.True(t, IsValidPort(65535)) + require.False(t, IsValidPort(65536)) +} + +func TestIsBindablePort(t *testing.T) { + t.Parallel() + + require.True(t, IsBindablePort(0)) + require.True(t, IsBindablePort(1)) + require.True(t, IsBindablePort(65535)) + require.False(t, IsBindablePort(65536)) +}