package api import ( "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "github.com/distribution/reference" "github.com/psviderski/uncloud/internal/machine/api/pb" "maps" "reflect" "regexp" "slices" ) const ( ServiceModeReplicated = "replicated" ServiceModeGlobal = "global" // PullPolicyAlways means the latest image is always pulled from the registry. PullPolicyAlways = "always" // PullPolicyMissing means the latest image is pulled from the registry only if it's not available in the cluster // (missing on all machines). If the image is available on any machine, its registry digest is used which may result // in pulling the image on machines where it's not available. This is the default pull policy. PullPolicyMissing = "missing" // PullPolicyNever means the image is never pulled from the registry. A service with this pull policy can only be // deployed to machines where the image is already available. PullPolicyNever = "never" ) var serviceIDRegexp = regexp.MustCompile("^[0-9a-f]{32}$") func ValidateServiceID(id string) bool { return serviceIDRegexp.MatchString(id) } type ServiceSpec struct { Container ContainerSpec // Mode is the replication mode of the service. Default is ServiceModeReplicated if empty. Mode string Name string // Ports defines what service ports to publish to make the service accessible outside the cluster. Ports []PortSpec // Replicas is the number of containers to run for the service. Only valid for a replicated service. Replicas uint } func (s *ServiceSpec) Validate() error { if err := s.Container.Validate(); err != nil { return err } switch s.Mode { case "", ServiceModeGlobal, ServiceModeReplicated: default: return fmt.Errorf("invalid mode: %q", s.Mode) } for _, p := range s.Ports { if (p.Mode == "" || p.Mode == PortModeIngress) && p.Protocol != ProtocolHTTP && p.Protocol != ProtocolHTTPS { return fmt.Errorf("unsupported protocol for ingress port %d: %s", p.ContainerPort, p.Protocol) } } // TODO: validate there is no conflict between ports. return nil } // ImmutableHash returns a hash of the immutable parts of the ServiceSpec that require container recreation if changed. func (s *ServiceSpec) ImmutableHash() (string, error) { var err error // Serialise and sort the ports to ensure the hash is consistent. ports := make([]string, len(s.Ports)) for i, p := range s.Ports { ports[i], err = p.String() if err != nil { return "", fmt.Errorf("encode service port spec: %w", err) } } slices.Sort(ports) volumes := make([]string, 0, len(s.Container.Volumes)) volumes = append(volumes, s.Container.Volumes...) slices.Sort(volumes) hashSpec := immutableHashSpec{ Command: s.Container.Command, Entrypoint: s.Container.Entrypoint, Image: s.Container.Image, Init: s.Container.Init, Ports: ports, Volumes: volumes, } data, err := json.Marshal(hashSpec) if err != nil { return "", fmt.Errorf("marshal immutable hash spec: %w", err) } hasher := sha256.New() if _, err = hasher.Write(data); err != nil { return "", fmt.Errorf("write to SHA256 hasher: %w", err) } return hex.EncodeToString(hasher.Sum(nil)), nil } // immutableHashSpec contains only the immutable fields from ServiceSpec that require container recreation if changed. type immutableHashSpec struct { Command []string `json:",omitempty"` Entrypoint []string `json:",omitempty"` Image string Init *bool `json:",omitempty"` // Ports are set as labels on the container which are immutable. // TODO: store ingress ports in the cluster store instead of as labels which will allow changing them without // recreating the container. Ports []string `json:",omitempty"` Volumes []string `json:",omitempty"` } // Equals returns true if the service spec is equal to the given spec ignoring the number of replicas. func (s *ServiceSpec) Equals(spec ServiceSpec) bool { // TODO: ignore order of ports. sCopy := *s // Ignore the number of replicas when comparing. sCopy.Replicas = 0 spec.Replicas = 0 return reflect.DeepEqual(*s, spec) } func (s *ServiceSpec) Clone() ServiceSpec { spec := *s if s.Ports != nil { spec.Ports = make([]PortSpec, len(s.Ports)) copy(spec.Ports, s.Ports) } spec.Container = s.Container.Clone() return spec } type ContainerSpec struct { // Command overrides the default CMD of the image to be executed when running a container. Command []string // Entrypoint overrides the default ENTRYPOINT of the image. Entrypoint []string Image string // Run a custom init inside the container. If nil, use the daemon's configured settings. Init *bool // PullPolicy determines when to pull the image from the registry or use the image already available in the cluster. // Default is PullPolicyMissing if empty. PullPolicy string // List of volumes to bind mount into the container. Volumes []string } func (s *ContainerSpec) Validate() error { if _, err := reference.ParseDockerRef(s.Image); err != nil { return fmt.Errorf("invalid image: %w", err) } return nil } func (s *ContainerSpec) Clone() ContainerSpec { spec := *s if s.Command != nil { spec.Command = make([]string, len(s.Command)) copy(spec.Command, s.Command) } if s.Entrypoint != nil { spec.Entrypoint = make([]string, len(s.Entrypoint)) copy(spec.Entrypoint, s.Entrypoint) } if s.Volumes != nil { spec.Volumes = make([]string, len(s.Volumes)) copy(spec.Volumes, s.Volumes) } return spec } type Service struct { ID string Name string Mode string Containers []MachineContainer } type MachineContainer struct { MachineID string Container Container } // Endpoints returns the exposed HTTP and HTTPS endpoints of the service. func (s *Service) Endpoints() []string { endpoints := make(map[string]struct{}) // Container specs may differ between containers in the same service, e.g. during a rolling update, // so we need to collect all unique endpoints. for _, ctr := range s.Containers { ports, err := ctr.Container.ServicePorts() if err != nil { continue } for _, port := range ports { protocol := "" switch port.Protocol { case ProtocolHTTP: protocol = "http" case ProtocolHTTPS: protocol = "https" default: continue } if port.Hostname == "" { // There shouldn't be http(s) ports without a hostname but just in case ignore them. continue } endpoint := fmt.Sprintf("%s://%s", protocol, port.Hostname) if port.PublishedPort != 0 { // For non-standard ports (80/443), include the port in the URL. if !(port.Protocol == ProtocolHTTP && port.PublishedPort == 80) && !(port.Protocol == ProtocolHTTPS && port.PublishedPort == 443) { endpoint += fmt.Sprintf(":%d", port.PublishedPort) } } endpoint += fmt.Sprintf(" → :%d", port.ContainerPort) endpoints[endpoint] = struct{}{} } } return slices.Sorted(maps.Keys(endpoints)) } func ServiceFromProto(s *pb.Service) (Service, error) { var err error containers := make([]MachineContainer, len(s.Containers)) for i, sc := range s.Containers { containers[i], err = machineContainerFromProto(sc) if err != nil { return Service{}, err } } return Service{ ID: s.Id, Name: s.Name, Mode: s.Mode, Containers: containers, }, nil } func machineContainerFromProto(sc *pb.Service_Container) (MachineContainer, error) { var c Container if err := json.Unmarshal(sc.Container, &c); err != nil { return MachineContainer{}, fmt.Errorf("unmarshal container: %w", err) } return MachineContainer{ MachineID: sc.MachineId, Container: c, }, nil }