Merge branch 'main' into smuu/20241204-retain-file-permission

pkg/instance/errors.go

@@ -218,6 +218,7 @@ var (
 	ErrApplyingFunctionToSidecar                 = errors.New("ApplyingFunctionToSidecar", "error applying function to sidecar '%s'")
 	ErrInitializingSidecar                       = errors.New("InitializingSidecar", "error initializing sidecar for instance '%s'")
 	ErrSidecarInstanceIsNil                      = errors.New("SidecarInstanceIsNil", "sidecar instance is nil for instance '%s'")
+	ErrFailedToCreatePersistentVolumeClaim       = errors.New("FailedToCreatePersistentVolumeClaim", "failed to create persistent volume claim")
 	ErrFailedToDeletePersistentVolumeClaim       = errors.New("FailedToDeletePersistentVolumeClaim", "failed to delete persistent volume claim")
 	ErrUpgradingImageNotAllowed                  = errors.New("UpgradingImageNotAllowed", "upgrading image is only allowed in state 'Started'. Current state is '%s'")
 	ErrAddingHostToProxyNotAllowed               = errors.New("AddingHostToProxyNotAllowed", "adding host to proxy is only allowed in state 'Started' and 'Preparing'. Current state is '%s'")
@@ -227,4 +228,7 @@ var (
 	ErrCannotCloneInstance                       = errors.New("CannotCloneInstance", "cannot clone instance '%s' in state '%s'")
 	ErrGettingIPNotAllowed                       = errors.New("GettingIPNotAllowed", "getting IP is allowed in state 'Started'. Current state is '%s'")
 	ErrPodIPNotReady                             = errors.New("PodIPNotReady", "pod IP is not ready for pod '%s'")
+	ErrFailedToGetFileSize                       = errors.New("FailedToGetFileSize", "failed to get file size")
+	ErrFileTooLargeCommitted                     = errors.New("FileTooLargeCommitted", "file '%s' is too large (max 1MiB) to add after instance is committed")
+	ErrTotalFilesSizeTooLarge                    = errors.New("TotalFilesSizeTooLarge", "total files size is too large (max 1MiB)")
 )

pkg/instance/execution.go

@@ -384,6 +384,8 @@ func (e *execution) prepareReplicaSetConfig() k8s.ReplicaSetConfig {
 		StartupProbe:    e.instance.monitoring.startupProbe,
 		Files:           e.instance.storage.files,
 		SecurityContext: e.instance.security.prepareSecurityContext(),
+		TCPPorts:        e.instance.network.portsTCP,
+		UDPPorts:        e.instance.network.portsUDP,
 	}
 
 	sidecarConfigs := make([]k8s.ContainerConfig, 0)
@@ -403,6 +405,8 @@ func (e *execution) prepareReplicaSetConfig() k8s.ReplicaSetConfig {
 			StartupProbe:    sidecar.Instance().monitoring.startupProbe,
 			Files:           sidecar.Instance().storage.files,
 			SecurityContext: sidecar.Instance().security.prepareSecurityContext(),
+			TCPPorts:        sidecar.Instance().network.portsTCP,
+			UDPPorts:        sidecar.Instance().network.portsUDP,
 		})
 	}
 

pkg/instance/proxy.go

@@ -27,7 +27,7 @@ func (n *network) AddHost(ctx context.Context, port int) (host string, err error
 		serviceName = n.instance.parentInstance.name
 	}
 
-	prefix := fmt.Sprintf("%s-%d", serviceName, port)
+	prefix := fmt.Sprintf("%s-%s-%d", n.instance.Scope, serviceName, port)
 	if err := n.instance.Proxy.AddHost(ctx, serviceName, prefix, port); err != nil {
 		return "", ErrAddingToProxy.WithParams(serviceName).Wrap(err)
 	}

pkg/instance/storage.go

@@ -16,6 +16,8 @@ import (
 	"github.com/celestiaorg/knuu/pkg/names"
 )
 
+const maxTotalFilesBytes = 1024 * 1024
+
 type storage struct {
 	instance *Instance
 	volumes  []*k8s.Volume
@@ -52,6 +54,13 @@ func (s *storage) AddFile(src string, dest string, chown string) error {
 		s.instance.build.addFileToBuilder(src, dest, chown)
 		return nil
 	case StateCommitted, StateStopped:
+		srcInfo, err := os.Stat(src)
+		if err != nil {
+			return ErrFailedToGetFileSize.Wrap(err)
+		}
+		if srcInfo.Size() > maxTotalFilesBytes {
+			return ErrFileTooLargeCommitted.WithParams(src)
+		}
 		return s.addFileToInstance(buildDirPath, dest, chown)
 	}
 
@@ -313,6 +322,29 @@ func (s *storage) addFileToInstance(srcPath, dest, chown string) error {
 
 	// get the permission of the src file
 	permission := fmt.Sprintf("%o", srcInfo.Mode().Perm())
+
+	size := int64(0)
+	for _, file := range s.files {
+		srcInfo, err := os.Stat(file.Source)
+		if err != nil {
+			return ErrFailedToGetFileSize.Wrap(err)
+		}
+		size += srcInfo.Size()
+	}
+	srcInfo, err = os.Stat(dstPath)
+	if err != nil {
+		return ErrFailedToGetFileSize.Wrap(err)
+	}
+	size += srcInfo.Size()
+	if size > maxTotalFilesBytes {
+		return ErrTotalFilesSizeTooLarge.WithParams(dstPath)
+	}
+
+	file := s.instance.K8sClient.NewFile(dstPath, dest)
+	parts := strings.Split(chown, ":")
+	if len(parts) != 2 {
+		return ErrInvalidFormat
+	}
 
 	file := s.instance.K8sClient.NewFile(srcPath, dest, chown, permission)
 
@@ -334,7 +366,10 @@ func (s *storage) deployVolume(ctx context.Context) error {
 	for _, volume := range s.volumes {
 		totalSize.Add(volume.Size)
 	}
-	s.instance.K8sClient.CreatePersistentVolumeClaim(ctx, s.instance.name, s.instance.execution.Labels(), totalSize)
+	err := s.instance.K8sClient.CreatePersistentVolumeClaim(ctx, s.instance.name, s.instance.execution.Labels(), totalSize)
+	if err != nil {
+		return ErrFailedToCreatePersistentVolumeClaim.Wrap(err)
+	}
 	s.instance.Logger.WithFields(logrus.Fields{
 		"total_size": totalSize.String(),
 		"instance":   s.instance.name,

pkg/k8s/pod.go

@@ -49,6 +49,8 @@ type ContainerConfig struct {
 	StartupProbe    *v1.Probe           // Startup probe for the container
 	Files           []*File             // Files to add to the Pod
 	SecurityContext *v1.SecurityContext // Security context for the container
+	TCPPorts        []int               // TCP ports to expose on the Pod
+	UDPPorts        []int               // UDP ports to expose on the Pod
 }
 
 type PodConfig struct {
@@ -553,6 +555,25 @@ func buildResources(memoryRequest, memoryLimit, cpuRequest resource.Quantity) v1
 	}
 }
 
+func buildPodPorts(tcpPorts, udpPorts []int) []v1.ContainerPort {
+	ports := make([]v1.ContainerPort, 0, len(tcpPorts)+len(udpPorts))
+	for _, port := range tcpPorts {
+		ports = append(ports, v1.ContainerPort{
+			Name:          fmt.Sprintf("tcp-%d", port),
+			Protocol:      v1.ProtocolTCP,
+			ContainerPort: int32(port),
+		})
+	}
+	for _, port := range udpPorts {
+		ports = append(ports, v1.ContainerPort{
+			Name:          fmt.Sprintf("udp-%d", port),
+			Protocol:      v1.ProtocolUDP,
+			ContainerPort: int32(port),
+		})
+	}
+	return ports
+}
+
 // prepareContainer creates a v1.Container from a given ContainerConfig.
 func prepareContainer(config ContainerConfig) v1.Container {
 	return v1.Container{
@@ -564,6 +585,7 @@ func prepareContainer(config ContainerConfig) v1.Container {
 		Env:             buildEnv(config.Env),
 		VolumeMounts:    buildContainerVolumes(config.Name, config.Volumes, config.Files),
 		Resources:       buildResources(config.MemoryRequest, config.MemoryLimit, config.CPURequest),
+		Ports:           buildPodPorts(config.TCPPorts, config.UDPPorts),
 		LivenessProbe:   config.LivenessProbe,
 		ReadinessProbe:  config.ReadinessProbe,
 		StartupProbe:    config.StartupProbe,

pkg/traefik/traefik.go

@@ -17,7 +17,6 @@ import (
 	"k8s.io/utils/ptr"
 
 	"github.com/celestiaorg/knuu/pkg/k8s"
-	"github.com/celestiaorg/knuu/pkg/names"
 )
 
 const (
@@ -26,6 +25,7 @@ const (
 	Port                   = 80
 	PortSecure             = 443
 	deploymentName         = "traefik-deployment"
+	serviceAccountName     = "traefik-service-account"
 	roleName               = "traefik-role"
 	containerName          = "traefik"
 	image                  = "traefik:v3.0"
@@ -51,22 +51,14 @@ func (t *Traefik) Deploy(ctx context.Context) error {
 		return ErrTraefikClientNotInitialized
 	}
 
-	// Create a dedicated service account for Traefik
-	serviceAccountName, err := names.NewRandomK8("traefik-service-account")
-	if err != nil {
-		return err
-	}
 	if err := t.K8sClient.CreateServiceAccount(ctx, serviceAccountName, nil); err != nil {
 		return ErrFailedToCreateServiceAccount.Wrap(err)
 	}
 
-	clusterRoleName, err := names.NewRandomK8(roleName)
-	if err != nil {
-		return err
-	}
+	clusterRoleName := k8s.SanitizeName(t.K8sClient.Namespace() + "-" + roleName)
 
 	// Define and create a ClusterRole for Traefik
-	err = t.K8sClient.CreateClusterRole(ctx, clusterRoleName, nil, []rbacv1.PolicyRule{
+	err := t.K8sClient.CreateClusterRole(ctx, clusterRoleName, nil, []rbacv1.PolicyRule{
 		{
 			APIGroups: []string{""}, // Core group
 			Resources: []string{"pods", "endpoints", "secrets", "services"},
@@ -209,10 +201,7 @@ func (t *Traefik) Endpoint(ctx context.Context) (string, error) {
 }
 
 func (t *Traefik) AddHost(ctx context.Context, serviceName, prefix string, portTCP int) error {
-	middlewareName, err := names.NewRandomK8("strip-" + prefix)
-	if err != nil {
-		return ErrGeneratingRandomK8sName.Wrap(err)
-	}
+	middlewareName := k8s.SanitizeName(prefix + "-strip")
 
 	// middleware is required to strip the prefix from the service name
 	if err := t.createMiddleware(ctx, prefix, middlewareName); err != nil {
@@ -303,11 +292,7 @@ func (t *Traefik) createIngressRoute(
 		Resource: "ingressroutes",
 	}
 
-	ingressRouteName, err := names.NewRandomK8("ing-route-" + prefix)
-	if err != nil {
-		return ErrTraefikIngressRouteCreationFailed.Wrap(err)
-	}
-
+	ingressRouteName := k8s.SanitizeName(prefix + "-ing-route")
 	ingressRoute := &unstructured.Unstructured{
 		Object: map[string]interface{}{
 			"apiVersion": "traefik.io/v1alpha1",
@@ -341,7 +326,7 @@ func (t *Traefik) createIngressRoute(
 		},
 	}
 
-	_, err = t.K8sClient.DynamicClient().Resource(ingressRouteGVR).Namespace(t.K8sClient.Namespace()).
+	_, err := t.K8sClient.DynamicClient().Resource(ingressRouteGVR).Namespace(t.K8sClient.Namespace()).
 		Create(ctx, ingressRoute, metav1.CreateOptions{})
 	if err != nil {
 		return ErrTraefikIngressRouteCreationFailed.Wrap(err)