Remove overwriting the default HTTP transport

docs/content/en/docs/5_architecture_and_components/function-runner/functionality/image-registry-management.md

@@ -207,7 +207,7 @@ The Function Runner supports custom TLS certificates for secure registry connect
 
 **Certificate loading:**
 - Reads from mounted secret path
-- Supports ca.crt or ca.pem filenames
+- Supports common certificate filenames
 - Parses PEM-encoded certificates
 - Creates certificate pool with custom CA
 

docs/content/en/docs/6_configuration_and_deployments/configurations/components/function-runner-config/private-registries-config.md

@@ -125,7 +125,12 @@ type: kubernetes.io/tls
 ```
 
 {{% alert title="Note" color="primary" %}}
-The certificate must be in PEM format and the key must be named `ca.crt` or `ca.pem`.
+The certificate must be in PEM format and the key must be named one of the following:
+- `ca.crt`
+- `ca.pem`
+- `cacert.pem`
+- `ca-bundle.crt`
+- `root.crt`
 {{% /alert %}}
 
 ### 2. Mount TLS Secret

func/internal/executableevaluator_test.go

@@ -37,6 +37,7 @@ const (
 	defaultKRMImagePrefix = "ghcr.io/kptdev/krm-functions-catalog/"
 	setImageFunction      = "set-image"
 	starlarkFunction      = "starlark"
+	testImageName         = "test-image"
 )
 
 func getFunctionConfigStore(binaryDir string) *functionconfigs.FunctionConfigStore {

func/internal/podevaluator.go

@@ -135,39 +135,44 @@ func NewPodEvaluator(ctx context.Context, o PodEvaluatorOptions, cl client.Clien
 	readyCh := make(chan *podReadyResponse, channelBufferSize)
 	evictCh := make(chan *podEvictionRequest, channelBufferSize)
 
+	podMgr := &podManager{
+		kubeClient:         cl,
+		namespace:          o.PodNamespace,
+		wrapperServerImage: o.WrapperServerImage,
+		podReadyCh:         readyCh,
+		podReadyTimeout:    60 * time.Second,
+		managerNamespace:   managerNs,
+		maxGrpcMessageSize: o.MaxGrpcMessageSize,
+
+		enablePrivateRegistries:    o.EnablePrivateRegistries,
+		registryAuthSecretPath:     o.RegistryAuthSecretPath,
+		registryAuthSecretName:     o.RegistryAuthSecretName,
+		enablePrivateRegistriesTls: o.EnablePrivateRegistriesTls,
+		tlsSecretPath:              o.TlsSecretPath,
+		tagResolver:                runtime.TagResolver{}, // TODO: no resolvers, kpt needs to expose these better
+	}
+
+	pcm := &podCacheManager{
+		gcScanInterval:             o.GcScanInterval,
+		podTTL:                     o.PodTTL,
+		connectionRequestCh:        reqCh,
+		podReadyCh:                 readyCh,
+		evictionCh:                 evictCh,
+		functions:                  map[string]*functionInfo{},
+		maxWaitlistLength:          maxWaitlist,
+		maxParallelPodsPerFunction: maxPods,
+		functionConfigMap:          functionConfigStore,
+
+		podManager: podMgr,
+	}
+
 	pe := &podEvaluator{
-		requestCh:      reqCh,
-		evictionCh:     evictCh,
-		maxGrpcRetries: maxRetries,
-		podCacheManager: &podCacheManager{
-			gcScanInterval:             o.GcScanInterval,
-			podTTL:                     o.PodTTL,
-			connectionRequestCh:        reqCh,
-			podReadyCh:                 readyCh,
-			evictionCh:                 evictCh,
-			functions:                  map[string]*functionInfo{},
-			maxWaitlistLength:          maxWaitlist,
-			maxParallelPodsPerFunction: maxPods,
-			functionConfigMap:          functionConfigStore,
-
-			podManager: &podManager{
-				kubeClient:         cl,
-				namespace:          o.PodNamespace,
-				wrapperServerImage: o.WrapperServerImage,
-				podReadyCh:         readyCh,
-				podReadyTimeout:    60 * time.Second,
-				managerNamespace:   managerNs,
-				maxGrpcMessageSize: o.MaxGrpcMessageSize,
-
-				enablePrivateRegistries:    o.EnablePrivateRegistries,
-				registryAuthSecretPath:     o.RegistryAuthSecretPath,
-				registryAuthSecretName:     o.RegistryAuthSecretName,
-				enablePrivateRegistriesTls: o.EnablePrivateRegistriesTls,
-				tlsSecretPath:              o.TlsSecretPath,
-				tagResolver:                runtime.TagResolver{},
-			},
-		},
+		requestCh:       reqCh,
+		evictionCh:      evictCh,
+		maxGrpcRetries:  maxRetries,
+		podCacheManager: pcm,
 	}
+
 	go pe.podCacheManager.podCacheManager(ctx)
 
 	err = pe.podCacheManager.retrieveFunctionPods(context.Background())

func/internal/podevaluator_tag_resolution_test.go

@@ -32,10 +32,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-const (
-	testImageName = "test-image"
-)
-
 type fakeLister struct {
 	tags map[string][]string
 	err  string

func/internal/podevaluator_unit_test.go

@@ -23,6 +23,8 @@ import (
 	"time"
 
 	"github.com/kptdev/kpt/pkg/fn/runtime"
+	"github.com/kptdev/kpt/pkg/lib/runneroptions"
+	fnconf "github.com/kptdev/porch/controllers/functionconfigs"
 	pb "github.com/kptdev/porch/func/evaluator"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -31,6 +33,7 @@ import (
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 // startFakeEvalServer starts a gRPC function evaluator server on a dynamic port.
@@ -50,6 +53,29 @@ func startFakeEvalServer(t *testing.T, evalFunc func(ctx context.Context, req *p
 	}
 }
 
+func TestNewPodEvaluator(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	kubeClient := fake.NewClientBuilder().Build()
+	store := fnconf.NewFunctionConfigStore(runneroptions.GHCRImagePrefix, "/functions")
+
+	eval, err := NewPodEvaluator(ctx, PodEvaluatorOptions{
+		PodNamespace:       "test-ns",
+		WrapperServerImage: "ghcr.io/kptdev/wrapper-server:latest",
+		GcScanInterval:     time.Minute,
+		PodTTL:             time.Minute,
+	}, kubeClient, store)
+	require.NoError(t, err)
+	require.NotNil(t, eval)
+
+	pe, ok := eval.(*podEvaluator)
+	require.True(t, ok)
+	assert.Equal(t, defaultMaxWaitlistLength, pe.podCacheManager.maxWaitlistLength)
+	assert.Equal(t, defaultMaxParallelPodsPerFunction, pe.podCacheManager.maxParallelPodsPerFunction)
+	assert.Equal(t, defaultMaxGrpcRetries, pe.maxGrpcRetries)
+}
+
 func TestEvaluateFunction_ErrorInResponse(t *testing.T) {
 	reqCh := make(chan *connectionRequest, 1)
 	pe := &podEvaluator{requestCh: reqCh,

func/internal/podmanager.go

@@ -36,6 +36,8 @@ import (
 	"github.com/kptdev/kpt/pkg/fn/runtime"
 	configapi "github.com/kptdev/porch/api/porchconfig/v1alpha1"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"go.uber.org/multierr"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
@@ -481,43 +483,71 @@ func (pm *podManager) getImageMetadata(ctx context.Context, ref name.Reference,
 }
 
 func (pm *podManager) getImage(ctx context.Context, ref name.Reference, auth authn.Authenticator, image string) (containerregistry.Image, error) {
+	remoteOpts := []remote.Option{
+		remote.WithContext(ctx),
+		remote.WithAuth(auth),
+	}
+
+	nonTlsTransport := otelTransport(nil)
+
 	// if private registries or their appropriate tls configuration are disabled in the config we pull image with default operation otherwise try and use their tls cert's
 	if !pm.enablePrivateRegistries || strings.HasPrefix(image, defaultRegistry) || !pm.enablePrivateRegistriesTls {
-		return remote.Image(ref, remote.WithAuth(auth), remote.WithContext(ctx))
-	}
-	tlsFile := "ca.crt"
-	// Check if mounted secret location contains CA file.
-	if _, err := os.Stat(pm.tlsSecretPath); os.IsNotExist(err) {
-		return nil, err
-	}
-	if _, errCRT := os.Stat(filepath.Join(pm.tlsSecretPath, "ca.crt")); os.IsNotExist(errCRT) {
-		if _, errPEM := os.Stat(filepath.Join(pm.tlsSecretPath, "ca.pem")); os.IsNotExist(errPEM) {
-			return nil, fmt.Errorf("ca.crt not found: %v, and ca.pem also not found: %w", errCRT, errPEM)
-		}
-		tlsFile = "ca.pem"
+		return remote.Image(ref, append(remoteOpts, remote.WithTransport(nonTlsTransport))...)
 	}
-	// Load the custom TLS configuration
-	tlsConfig, err := loadTLSConfig(filepath.Join(pm.tlsSecretPath, tlsFile))
+
+	tlsTransport, err := makeTlsTransport(pm.tlsSecretPath)
 	if err != nil {
 		return nil, err
 	}
-	// Create a custom HTTPS transport
-	transport := createTransport(tlsConfig)
 
 	// Attempt image pull with given custom TLS cert
-	img, tlsErr := remote.Image(ref, remote.WithAuth(auth), remote.WithContext(ctx), remote.WithTransport(transport))
+	img, tlsErr := remote.Image(ref, append(remoteOpts, remote.WithTransport(tlsTransport))...)
 	if tlsErr != nil {
 		// Attempt without given custom TLS cert but with default keychain
 		klog.Errorf("Pulling image %s with the provided TLS Cert has failed with error %v", image, tlsErr)
 		klog.Infof("Attempting image pull with default keychain instead of provided TLS Cert")
-		img, err = remote.Image(ref, remote.WithAuth(auth), remote.WithContext(ctx))
+		var err error
+		img, err = remote.Image(ref, append(remoteOpts, remote.WithTransport(nonTlsTransport))...)
 		if err != nil {
 			return nil, fmt.Errorf("failed to pull image %s with default keychain: %w\n  (pull was retried after this TLS error: %v)", ref.String(), err, tlsErr)
 		}
 	}
 	return img, nil
 }
 
+func makeTlsTransport(tlsPath string) (http.RoundTripper, error) {
+	caCertPath, err := tlsCACertPath(tlsPath)
+	if err != nil {
+		return nil, err
+	}
+	tlsConfig, err := loadTLSConfig(caCertPath)
+	if err != nil {
+		return nil, err
+	}
+	return otelTransport(tlsConfig), nil
+}
+
+func tlsCACertPath(tlsSecretPath string) (string, error) {
+	if _, err := os.Stat(tlsSecretPath); err != nil {
+		return "", fmt.Errorf("tls secret folder %q could not be reached: %w", tlsSecretPath, err)
+	}
+
+	var multiErr error
+
+	candidates := []string{"ca.crt", "ca.pem", "cacert.pem", "ca-bundle.crt", "root.crt"}
+	for _, file := range candidates {
+		path := filepath.Join(tlsSecretPath, file)
+		if _, err := os.Stat(path); err == nil {
+			return path, nil
+		} else {
+			multierr.AppendInto(&multiErr, err)
+		}
+	}
+
+	return "", fmt.Errorf("no CA certificate found in %q (candidates: [%s]): %w",
+		tlsSecretPath, strings.Join(candidates, ", "), multiErr)
+}
+
 func loadTLSConfig(caCertPath string) (*tls.Config, error) {
 	// Read the CA certificate file
 	caCert, err := os.ReadFile(caCertPath)
@@ -537,10 +567,20 @@ func loadTLSConfig(caCertPath string) (*tls.Config, error) {
 	return tlsConfig, nil
 }
 
-func createTransport(tlsConfig *tls.Config) *http.Transport {
-	return &http.Transport{
-		TLSClientConfig: tlsConfig,
+// otelTransport returns an OpenTelemetry-instrumented transport.
+// tlsConfig is applied to the underlying transport when non-nil.
+func otelTransport(tlsConfig *tls.Config) http.RoundTripper {
+	if tlsConfig != nil {
+		defTransport, ok := http.DefaultTransport.(*http.Transport)
+		if !ok {
+			klog.Errorf("Cannot inject TLS into the default http transport as it has been replaced; will use a blank one")
+			defTransport = &http.Transport{}
+		}
+		baseTransport := defTransport.Clone()
+		baseTransport.TLSClientConfig = tlsConfig.Clone()
+		return otelhttp.NewTransport(baseTransport)
 	}
+	return otelhttp.NewTransport(http.DefaultTransport)
 }
 
 // CreatePod creates a pod for an image.