diff --git a/cmd/container_runtime/containerd.go b/cmd/container_runtime/containerd.go index f044110..48b0380 100644 --- a/cmd/container_runtime/containerd.go +++ b/cmd/container_runtime/containerd.go @@ -1,7 +1,6 @@ package runtime import ( - "context" "fmt" "os" "path/filepath" @@ -24,11 +23,6 @@ const ( DefaultConfigVersion = 2 ) -type ContainerdController interface { - Load(ctx context.Context, log *zerolog.Logger) (*registry.DefaultZotConfig, error) - Generate(ctx context.Context, configPath string, log *zerolog.Logger) error -} - var DefaultContainerDGenPath string func init() { @@ -49,7 +43,7 @@ func init() { func NewContainerdCommand() *cobra.Command { var generateConfig bool - var defaultZotConfig *registry.DefaultZotConfig + var defaultZotConfig registry.DefaultZotConfig var containerdConfigPath string var containerDCertPath string @@ -61,8 +55,8 @@ func NewContainerdCommand() *cobra.Command { }, RunE: func(cmd *cobra.Command, args []string) error { log := logger.FromContext(cmd.Context()) - sourceRegistry := config.GetRemoteRegistryURL() - satelliteHostConfig := NewSatelliteHostConfig(defaultZotConfig.GetLocalRegistryURL(), sourceRegistry) + sourceRegistry := config.GetSourceRegistryURL() + satelliteHostConfig := NewSatelliteHostConfig(defaultZotConfig.RemoteURL, sourceRegistry) if generateConfig { log.Info().Msg("Generating containerd config file for containerd ...") log.Info().Msgf("Fetching containerd config from path: %s", containerdConfigPath) @@ -71,7 +65,7 @@ func NewContainerdCommand() *cobra.Command { log.Err(err).Msg("Error generating containerd config") return fmt.Errorf("could not generate containerd config: %w", err) } - return GenerateContainerdConfig(defaultZotConfig, log, containerdConfigPath, containerDCertPath) + return GenerateContainerdConfig(log, containerdConfigPath, containerDCertPath) } return nil }, @@ -87,12 +81,17 @@ func NewContainerdCommand() *cobra.Command { // GenerateContainerdConfig generates the containerd config file for the containerd runtime // It takes the zot config a logger and the containerd config path // It reads the containerd config file and adds the local registry to the config file -func GenerateContainerdConfig(defaultZotConfig *registry.DefaultZotConfig, log *zerolog.Logger, containerdConfigPath, containerdCertPath string) error { +func GenerateContainerdConfig(log *zerolog.Logger, containerdConfigPath, containerdCertPath string) error { // First Read the present config file at the configPath data, err := utils.ReadFile(containerdConfigPath, false) if err != nil { - log.Err(err).Msg("Error reading config file") - return fmt.Errorf("could not read config file: %w", err) + if os.IsNotExist(err) { + log.Warn().Msg("Config file does not exist, proceeding with default values") + data = []byte{} + } else { + log.Err(err).Msg("Error reading config file") + return fmt.Errorf("could not read config file: %w", err) + } } // Now we marshal the data into the containerd config containerdConfig := &ContainerdConfigToml{} diff --git a/cmd/container_runtime/crio.go b/cmd/container_runtime/crio.go index 1dbe1a2..f355663 100644 --- a/cmd/container_runtime/crio.go +++ b/cmd/container_runtime/crio.go @@ -2,6 +2,7 @@ package runtime import ( "fmt" + "net/url" "os" "path/filepath" @@ -45,7 +46,7 @@ func NewCrioCommand() *cobra.Command { Use: "crio", Short: "Creates the config file for the crio runtime to fetch the images from the local repository", PersistentPreRunE: func(cmd *cobra.Command, args []string) error { - return SetupContainerRuntimeCommand(cmd, &defaultZotConfig, DefaultCrioGenPath) + return SetupContainerRuntimeCommand(cmd, defaultZotConfig, DefaultCrioGenPath) }, RunE: func(cmd *cobra.Command, args []string) error { log := logger.FromContext(cmd.Context()) @@ -71,7 +72,13 @@ func GenerateCrioRegistryConfig(defaultZotConfig *registry.DefaultZotConfig, cri // Read the current crio registry config file data, err := utils.ReadFile(crioConfigPath, false) if err != nil { - return fmt.Errorf("could not read crio registry config file: %w", err) + if os.IsNotExist(err) { + log.Warn().Msg("Config file does not exist, proceeding with default values") + data = []byte{} + } else { + log.Err(err).Msg("Error reading config file") + return fmt.Errorf("could not read config file: %w", err) + } } var crioRegistryConfig CriORegistryConfig err = toml.Unmarshal(data, &crioRegistryConfig) @@ -82,7 +89,7 @@ func GenerateCrioRegistryConfig(defaultZotConfig *registry.DefaultZotConfig, cri // Update the crio registry config file // - Add the local registry to the unqualified search registries if not already present var found bool = false - var localRegistry string = utils.FormatRegistryURL(defaultZotConfig.GetLocalRegistryURL()) + var localRegistry string = utils.FormatRegistryURL(defaultZotConfig.RemoteURL) for _, registry := range crioRegistryConfig.UnqualifiedSearchRegistries { if registry == localRegistry { found = true @@ -141,42 +148,27 @@ func GenerateCrioRegistryConfig(defaultZotConfig *registry.DefaultZotConfig, cri return nil } -func SetupContainerRuntimeCommand(cmd *cobra.Command, defaultZotConfig **registry.DefaultZotConfig, defaultGenPath string) error { +func SetupContainerRuntimeCommand(cmd *cobra.Command, defaultZotConfig *registry.DefaultZotConfig, defaultGenPath string) error { + utils.CommandRunSetup(cmd) var err error - checks, warnings := config.InitConfig(config.DefaultConfigPath) - if len(checks) > 0 || len(warnings) > 0 { - log := logger.FromContext(cmd.Context()) - for _, warn := range warnings { - log.Warn().Msg(warn) - } - for _, err := range checks { - log.Error().Err(err).Msg("Error initializing config") - } - return fmt.Errorf("error initializing config") - } - if err != nil { - return fmt.Errorf("could not initialize config: %w", err) - } - utils.SetupContextForCommand(cmd) log := logger.FromContext(cmd.Context()) if config.GetOwnRegistry() { log.Info().Msg("Using own registry for config generation") - address, err := utils.ValidateRegistryAddress(config.GetOwnRegistryAdr(), config.GetOwnRegistryPort()) + log.Info().Msgf("Remote registry URL: %s", config.GetRemoteRegistryURL()) + _, err := url.Parse(config.GetRemoteRegistryURL()) if err != nil { - log.Err(err).Msg("Error validating registry address") - return err + return fmt.Errorf("could not parse remote registry URL: %w", err) } - log.Info().Msgf("Registry address validated: %s", address) - (*defaultZotConfig).HTTP.Address = config.GetOwnRegistryAdr() - (*defaultZotConfig).HTTP.Port = config.GetOwnRegistryPort() + defaultZotConfig.RemoteURL = config.GetRemoteRegistryURL() } else { log.Info().Msg("Using default registry for config generation") - *defaultZotConfig, err = registry.ReadConfig(config.GetZotConfigPath()) - if err != nil || *defaultZotConfig == nil { + defaultZotConfig, err = registry.ReadConfig(config.GetZotConfigPath()) + if err != nil || defaultZotConfig == nil { return fmt.Errorf("could not read config: %w", err) } - log.Info().Msgf("Default config read successfully: %v", (*defaultZotConfig).HTTP.Address+":"+(*defaultZotConfig).HTTP.Port) + defaultZotConfig.RemoteURL = defaultZotConfig.GetLocalRegistryURL() + log.Info().Msgf("Default config read successfully: %v", defaultZotConfig.HTTP.Address+":"+defaultZotConfig.HTTP.Port) } return utils.CreateRuntimeDirectory(defaultGenPath) } diff --git a/cmd/container_runtime/read_config.go b/cmd/container_runtime/read_config.go index 8144f2e..53612a7 100644 --- a/cmd/container_runtime/read_config.go +++ b/cmd/container_runtime/read_config.go @@ -2,9 +2,7 @@ package runtime import ( "fmt" - "os" - "container-registry.com/harbor-satellite/internal/config" "container-registry.com/harbor-satellite/internal/utils" "container-registry.com/harbor-satellite/logger" "github.com/spf13/cobra" @@ -14,23 +12,8 @@ func NewReadConfigCommand(runtime string) *cobra.Command { readContainerdConfig := &cobra.Command{ Use: "read", Short: fmt.Sprintf("Reads the config file for the %s runtime", runtime), - PersistentPreRun: func(cmd *cobra.Command, args []string) { - checks, warnings := config.InitConfig(config.DefaultConfigPath) - if len(checks) > 0 || len(warnings) > 0 { - ctx := cmd.Context() - ctx, cancel := utils.SetupContext(ctx) - ctx = logger.AddLoggerToContext(ctx, "info") - log := logger.FromContext(ctx) - for _, warn := range warnings { - log.Warn().Msg(warn) - } - for _, err := range checks { - log.Error().Err(err).Msg("Error initializing config") - } - cancel() - os.Exit(1) - } - utils.SetupContextForCommand(cmd) + PersistentPreRunE: func(cmd *cobra.Command, args []string) error { + return utils.CommandRunSetup(cmd) }, RunE: func(cmd *cobra.Command, args []string) error { //Parse the flags diff --git a/cmd/root.go b/cmd/root.go index 574e510..814ca68 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -2,7 +2,6 @@ package cmd import ( "context" - "os" runtime "container-registry.com/harbor-satellite/cmd/container_runtime" "container-registry.com/harbor-satellite/internal/config" @@ -20,29 +19,12 @@ func NewRootCommand() *cobra.Command { rootCmd := &cobra.Command{ Use: "harbor-satellite", Short: "Harbor Satellite is a tool to replicate images from source registry to Harbor registry", - PersistentPreRun: func(cmd *cobra.Command, args []string) { - errors, warnings := config.InitConfig(config.DefaultConfigPath) - if len(errors) > 0 || len(warnings) > 0 { - ctx := cmd.Context() - ctx, cancel := utils.SetupContext(ctx) - ctx = logger.AddLoggerToContext(ctx, "info") - log := logger.FromContext(ctx) - for _, warn := range warnings { - log.Warn().Msg(warn) - } - for _, err := range errors { - log.Error().Err(err).Msg("Error initializing config") - } - if len(errors) > 0 { - cancel() - os.Exit(1) - } - } + PersistentPreRunE: func(cmd *cobra.Command, args []string) error { + return utils.CommandRunSetup(cmd) }, RunE: func(cmd *cobra.Command, args []string) error { ctx := cmd.Context() ctx, cancel := utils.SetupContext(ctx) - ctx = logger.AddLoggerToContext(ctx, config.GetLogLevel()) return run(ctx, cancel) }, } @@ -75,15 +57,20 @@ func run(ctx context.Context, cancel context.CancelFunc) error { log.Error().Err(err).Msg("Error starting scheduler") return err } - - satelliteService := satellite.NewSatellite(ctx, scheduler.GetSchedulerKey()) + localRegistryConfig := satellite.NewRegistryConfig(config.GetRemoteRegistryURL(), config.GetRemoteRegistryUsername(), config.GetRemoteRegistryPassword()) + sourceRegistryConfig := satellite.NewRegistryConfig(config.GetSourceRegistryURL(), config.GetSourceRegistryUsername(), config.GetSourceRegistryPassword()) + states := config.GetStates() + useUnsecure := config.UseUnsecure() + satelliteService := satellite.NewSatellite(ctx, scheduler.GetSchedulerKey(), localRegistryConfig, sourceRegistryConfig, useUnsecure, states) g.Go(func() error { return satelliteService.Run(ctx) }) log.Info().Msg("Startup complete 🚀") - return g.Wait() + g.Wait() + scheduler.Stop() + return nil } func setupServerApp(ctx context.Context, log *zerolog.Logger) *server.App { diff --git a/config.json b/config.json index d4135f7..9a57110 100644 --- a/config.json +++ b/config.json @@ -1,34 +1,39 @@ { + "state_config": { + "auth": { + "name": "admin", + "registry": "https://registry.bupd.xyz", + "secret": "Harbor12345" + }, + "states": [ + "https://registry.bupd.xyz/satellite/mehulsat/state" + ] + }, "environment_variables": { - "bring_own_registry": false, "ground_control_url": "https://gc.bupd.xyz", "log_level": "info", - "own_registry_adr": "127.0.0.1", - "own_registry_port": "8585", - "zot_config_path": "./registry/config.json", "use_unsecure": true, - "token": "187916a603c135d681ef0553752994f3", - "state_fetch_period": "", - "config_fetch_period": "", + "zot_config_path": "./registry/config.json", + "token": "", "jobs": [ { "name": "replicate_state", - "seconds": "10", - "minutes": "*", - "hours": "*", - "day_of_month": "*", - "month": "*", - "day_of_week": "*" + "schedule": "@every 00h00m10s" }, { "name": "update_config", - "seconds": "5", - "minutes": "asdfasdfsaf", - "hours": "asdfas", - "day_of_month": "*", - "month": "*", - "day_of_week": "dsffa" + "schedule": "@every 00h00m30s" + }, + { + "name": "register_satellite", + "schedule": "@every 00h00m05s" } - ] + ], + "local_registry": { + "url": "http://127.0.0.1:8585", + "username": "methylisocyanide", + "password": "Harbor12345", + "bring_own_registry": false + } } -} +} \ No newline at end of file diff --git a/internal/config/config.go b/internal/config/config.go index c4e64ce..c82692b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -2,34 +2,37 @@ package config import ( "encoding/json" + "os" ) var appConfig *Config -const DefaultConfigPath string = "config.json" -const ReplicateStateJobName string = "replicate_state" -const UpdateConfigJobName string = "update_config" +// Warning represents a non-critical issue with configuration. +type Warning string type Auth struct { - Name string `json:"name,omitempty"` - Registry string `json:"registry,omitempty"` - Secret string `json:"secret,omitempty"` + SourceUsername string `json:"name,omitempty"` + Registry string `json:"registry,omitempty"` + SourcePassword string `json:"secret,omitempty"` +} + +type LocalRegistryConfig struct { + URL string `json:"url"` + UserName string `json:"username"` + Password string `json:"password"` + BringOwnRegistry bool `json:"bring_own_registry"` } // LocalJsonConfig is a struct that holds the configs that are passed as environment variables type LocalJsonConfig struct { - BringOwnRegistry bool `json:"bring_own_registry"` - GroundControlURL string `json:"ground_control_url"` - LogLevel string `json:"log_level"` - OwnRegistryAddr string `json:"own_registry_addr"` - OwnRegistryPort string `json:"own_registry_port"` - UseUnsecure bool `json:"use_unsecure"` - ZotConfigPath string `json:"zot_config_path"` - Token string `json:"token"` - StateFetchPeriod string `json:"state_fetch_period"` - ConfigFetchPeriod string `json:"config_fetch_period"` - Jobs []Job `json:"jobs"` + GroundControlURL string `json:"ground_control_url"` + LogLevel string `json:"log_level"` + UseUnsecure bool `json:"use_unsecure"` + ZotConfigPath string `json:"zot_config_path"` + Token string `json:"token"` + Jobs []Job `json:"jobs"` + LocalRegistryConfig LocalRegistryConfig `json:"local_registry"` } type StateConfig struct { @@ -38,100 +41,17 @@ type StateConfig struct { } type Config struct { - StateConfig StateConfig `json:"state_config"` - LocalJsonConfig LocalJsonConfig `json:"environment_variables"` - ZotUrl string `json:"zot_url"` + StateConfig StateConfig `json:"state_config,omitempty"` + LocalJsonConfig LocalJsonConfig `json:"environment_variables,omitempty"` + ZotUrl string `json:"zot_url,omitempty"` } type Job struct { - Name string `json:"name"` - Seconds string `json:"seconds"` - Minutes string `json:"minutes"` - Hours string `json:"hours"` - DayOfMonth string `json:"day_of_month"` - Month string `json:"month"` - DayOfWeek string `json:"day_of_week"` -} - -func GetLogLevel() string { - if appConfig.LocalJsonConfig.LogLevel == "" { - return "info" - } - return appConfig.LocalJsonConfig.LogLevel -} - -func GetOwnRegistry() bool { - return appConfig.LocalJsonConfig.BringOwnRegistry -} - -func GetOwnRegistryAdr() string { - return appConfig.LocalJsonConfig.OwnRegistryAddr -} - -func GetOwnRegistryPort() string { - return appConfig.LocalJsonConfig.OwnRegistryPort -} - -func GetZotConfigPath() string { - return appConfig.LocalJsonConfig.ZotConfigPath -} - -func GetInput() string { - return "" -} - -func SetZotURL(url string) { - appConfig.ZotUrl = url -} - -func GetZotURL() string { - return appConfig.ZotUrl -} - -func UseUnsecure() bool { - return appConfig.LocalJsonConfig.UseUnsecure -} - -func GetHarborPassword() string { - return appConfig.StateConfig.Auth.Secret -} - -func GetHarborUsername() string { - return appConfig.StateConfig.Auth.Name -} - -func SetRemoteRegistryURL(url string) { - appConfig.StateConfig.Auth.Registry = url -} - -func GetRemoteRegistryURL() string { - return appConfig.StateConfig.Auth.Registry -} - -func GetStateFetchPeriod() string { - return appConfig.LocalJsonConfig.StateFetchPeriod -} - -func GetConfigFetchPeriod() string { - return appConfig.LocalJsonConfig.ConfigFetchPeriod -} - -func GetStates() []string { - return appConfig.StateConfig.States -} - -func GetToken() string { - return appConfig.LocalJsonConfig.Token -} - -func GetGroundControlURL() string { - return appConfig.LocalJsonConfig.GroundControlURL -} - -func SetGroundControlURL(url string) { - appConfig.LocalJsonConfig.GroundControlURL = url + Name string `json:"name"` + Schedule string `json:"schedule"` } +// ParseConfigFromJson parses a JSON string into a Config struct. Returns an error if the JSON is invalid func ParseConfigFromJson(jsonData string) (*Config, error) { var config Config err := json.Unmarshal([]byte(jsonData), &config) @@ -141,6 +61,7 @@ func ParseConfigFromJson(jsonData string) (*Config, error) { return &config, nil } +// ReadConfigData reads the data from the specified path. Returns an error if the file does not exist or is a directory func ReadConfigData(configPath string) ([]byte, error) { fileInfo, err := os.Stat(configPath) @@ -157,9 +78,17 @@ func ReadConfigData(configPath string) ([]byte, error) { return data, nil } -func LoadConfig(configPath string) (*Config, []error, []string) { +// LoadConfig reads the configuration file from the specified path and returns a Config struct. Returns an error if the file does not exist or is a directory. +// Also returns a slice of errors and warnings if the configuration is invalid +// For jobs, we will do the following: +// 1. Check the jobs provided by the user in the config.json. +// 2. Validate the jobs provided by the user. +// 3. If the job cron schedule is not valid, set the default schedule and replace it in the jobs. +// 4. Once the job is validated, append it to the validJobs slice if the job name is valid, i.e., it is used by the satellite. +// 5. Finally, check for critical jobs that are not present in the config and manually add them to the validJobs slice. +func LoadConfig(configPath string) (*Config, []error, []Warning) { var checks []error - var warnings []string + var warnings []Warning var err error configData, err := ReadConfigData(configPath) if err != nil { @@ -172,22 +101,47 @@ func LoadConfig(configPath string) (*Config, []error, []string) { return nil, checks, warnings } // Validate the job schedule fields + var validJobs []Job for i := range config.LocalJsonConfig.Jobs { - warnings = append(warnings, ValidateJobSchedule(&config.LocalJsonConfig.Jobs[i])...) + if warning, err := ValidateCronJob(&config.LocalJsonConfig.Jobs[i]); err != nil { + checks = append(checks, err) + } else { + if warning != "" { + warnings = append(warnings, warning) + } + validJobs = append(validJobs, config.LocalJsonConfig.Jobs[i]) + } } + // Add essential jobs if they are not present + AddEssentialJobs(&validJobs) + config.LocalJsonConfig.Jobs = validJobs return config, checks, warnings } -func InitConfig(configPath string) ([]error, []string) { +// InitConfig reads the configuration file from the specified path and initializes the global appConfig variable. +func InitConfig(configPath string) ([]error, []Warning) { var err []error - var warnings []string + var warnings []Warning appConfig, err, warnings = LoadConfig(configPath) + WriteConfig(configPath) return err, warnings } -func UpdateStateConfig(name, registry, secret string, states []string) { - appConfig.StateConfig.Auth.Name = name +func UpdateStateAuthConfig(name, registry, secret string, states []string) { + appConfig.StateConfig.Auth.SourceUsername = name appConfig.StateConfig.Auth.Registry = registry - appConfig.StateConfig.Auth.Secret = secret + appConfig.StateConfig.Auth.SourcePassword = secret appConfig.StateConfig.States = states + WriteConfig(DefaultConfigPath) +} +func WriteConfig(configPath string) error { + data, err := json.MarshalIndent(appConfig, "", " ") + if err != nil { + return err + } + err = os.WriteFile(configPath, data, 0644) + if err != nil { + return err + } + return nil } diff --git a/internal/config/constants.go b/internal/config/constants.go new file mode 100644 index 0000000..8ce7c37 --- /dev/null +++ b/internal/config/constants.go @@ -0,0 +1,22 @@ +package config + +// Job names that the user is expected to provide in the config.json file +const ReplicateStateJobName string = "replicate_state" +const UpdateConfigJobName string = "update_config" +const ZTRConfigJobName string = "register_satellite" + +// The values below contain the default values of the constants used in the satellite. The user is allowed to override them +// by providing values in the config.json file. These default values will be used if the user does not provide any value or wrong format value +// in the config.json file. + +// Default config.json path for the satellite, used if the user does not provide any config path +const DefaultConfigPath string = "config.json" +const DefaultZotConfigPath string = "./registry/config.json" + +// Below are the default values of the job schedules that would be used if the user does not provide any schedule or +// if there is any error while parsing the cron expression +const DefaultFetchConfigFromGroundControlTimePeriod string = "@every 00h00m30s" +const DefaultZeroTouchRegistrationCronExpr string = "@every 00h00m05s" +const DefaultFetchAndReplicateStateTimePeriod string = "@every 00h00m10s" + +const BringOwnRegistry bool = false diff --git a/internal/config/getters.go b/internal/config/getters.go new file mode 100644 index 0000000..766ff96 --- /dev/null +++ b/internal/config/getters.go @@ -0,0 +1,73 @@ +package config + +func GetLogLevel() string { + if appConfig == nil || appConfig.LocalJsonConfig.LogLevel == "" { + return "info" + } + return appConfig.LocalJsonConfig.LogLevel +} + +func GetOwnRegistry() bool { + return appConfig.LocalJsonConfig.LocalRegistryConfig.BringOwnRegistry +} + +func GetZotConfigPath() string { + return appConfig.LocalJsonConfig.ZotConfigPath +} + +func SetRemoteRegistryURL(url string) { + appConfig.LocalJsonConfig.LocalRegistryConfig.URL = url + WriteConfig(DefaultConfigPath) +} + +func GetZotURL() string { + return appConfig.ZotUrl +} + +func UseUnsecure() bool { + return appConfig.LocalJsonConfig.UseUnsecure +} + +func GetSourceRegistryPassword() string { + return appConfig.StateConfig.Auth.SourcePassword +} + +func GetSourceRegistryUsername() string { + return appConfig.StateConfig.Auth.SourceUsername +} + +func SetSourceRegistryURL(url string) { + appConfig.StateConfig.Auth.Registry = url +} + +func GetSourceRegistryURL() string { + return appConfig.StateConfig.Auth.Registry +} + +func GetStates() []string { + return appConfig.StateConfig.States +} + +func GetToken() string { + return appConfig.LocalJsonConfig.Token +} + +func GetGroundControlURL() string { + return appConfig.LocalJsonConfig.GroundControlURL +} + +func SetGroundControlURL(url string) { + appConfig.LocalJsonConfig.GroundControlURL = url +} + +func GetRemoteRegistryUsername() string { + return appConfig.LocalJsonConfig.LocalRegistryConfig.UserName +} + +func GetRemoteRegistryPassword() string { + return appConfig.LocalJsonConfig.LocalRegistryConfig.Password +} + +func GetRemoteRegistryURL() string { + return appConfig.LocalJsonConfig.LocalRegistryConfig.URL +} diff --git a/internal/config/jobs.go b/internal/config/jobs.go new file mode 100644 index 0000000..9132b03 --- /dev/null +++ b/internal/config/jobs.go @@ -0,0 +1,90 @@ +package config + +import ( + "fmt" + + "github.com/robfig/cron/v3" +) + +// JobConfig holds default schedule and validation settings for jobs. +type JobConfig struct { + DefaultSchedule string +} + +var essentialJobs = []string{ReplicateStateJobName, UpdateConfigJobName, ZTRConfigJobName} + +// Predefined job configurations. +var jobConfigs = map[string]JobConfig{ + ReplicateStateJobName: { + DefaultSchedule: DefaultFetchAndReplicateStateTimePeriod, + }, + UpdateConfigJobName: { + DefaultSchedule: DefaultFetchConfigFromGroundControlTimePeriod, + }, + ZTRConfigJobName: { + DefaultSchedule: DefaultZeroTouchRegistrationCronExpr, + }, +} + +// ValidateCronJob validates a job's configuration, ensuring it has a valid schedule. +// It sets a default schedule if none is provided and returns warnings for any issues. +func ValidateCronJob(job *Job) (Warning, error) { + config, exists := jobConfigs[job.Name] + if !exists { + return "", fmt.Errorf("unknown job name %s", job.Name) + } + + warning := ensureSchedule(job, config.DefaultSchedule) + if cronWarning := validateCronExpression(job.Schedule); cronWarning != "" { + // Reset to default schedule if validation fails. + job.Schedule = config.DefaultSchedule + return cronWarning, nil + } + + return warning, nil +} + +// ensureSchedule ensures the job has a schedule, using the default if none is provided. +func ensureSchedule(job *Job, defaultSchedule string) Warning { + if job.Schedule == "" { + job.Schedule = defaultSchedule + return Warning(fmt.Sprintf("no schedule provided for job %s, using default schedule %s", job.Name, defaultSchedule)) + } + return "" +} + +// validateCronExpression checks the validity of a cron expression. +func validateCronExpression(cronExpression string) Warning { + + if _, err := cron.ParseStandard(cronExpression); err != nil { + return Warning(fmt.Sprintf("error parsing cron expression: %v", err)) + } + return "" +} + +// AddEssentialJobs adds essential jobs to the provided slice if they are not already present. +func AddEssentialJobs(jobsPresent *[]Job) { + for _, jobName := range essentialJobs { + found := false + for _, job := range *jobsPresent { + if job.Name == jobName { + found = true + break + } + } + if !found { + defaultSchedule := jobConfigs[jobName].DefaultSchedule + *jobsPresent = append(*jobsPresent, Job{Name: jobName, Schedule: defaultSchedule}) + } + } +} + + +func GetJobSchedule(jobName string) (string, error) { + for _, job := range appConfig.LocalJsonConfig.Jobs { + if job.Name == jobName { + return job.Schedule, nil + } + } + return "", fmt.Errorf("job %s not found", jobName) +} diff --git a/internal/config/validations.go b/internal/config/validations.go deleted file mode 100644 index 3f3e24d..0000000 --- a/internal/config/validations.go +++ /dev/null @@ -1,72 +0,0 @@ -package config - -import ( - "fmt" - "regexp" -) - -// Validation rules -var ( - numericPattern = `^(?:[0-9]|[1-5][0-9])$|^[\*/,\-]+$` - hoursPattern = `^(?:[0-9]|1[0-9]|2[0-3])$|^[\*/,\-]+$` - dayOfMonthPattern = `^(?:[1-9]|[12][0-9]|3[01])$|^[\*/,\-?]+$` - monthPattern = `^(?:[1-9]|1[0-2]|JAN|FEB|MAR|APR|MAY|JUN|JUL|AUG|SEP|OCT|NOV|DEC)$|^[\*/,\-]+$` - dayOfWeekPattern = `^(?:[0-6]|SUN|MON|TUE|WED|THU|FRI|SAT)$|^[\*/,\-?]+$` - defaultValues = map[string]string{ - "Seconds": "0", - "Minutes": "0", - "Hours": "0", - "DayOfMonth": "1", - "Month": "1", - "DayOfWeek": "0", - } -) - -// ValidateJobSchedule validates the job schedule fields and returns a list of errors -// if any of the fields are invalid as per the cron format rules for each field type -// Seconds: 0-59 or */,-,? -// Minutes: 0-59 or */,-,? -// Hours: 0-23 or */,- -// DayOfMonth: 1-31 or */,-,? -// Month: 1-12 or JAN-DEC or */,- -// DayOfWeek: 0-6 or SUN-SAT or */,-,? -// It would also update the job schedule fields with default values if any of the fields are invalid -func ValidateJobSchedule(job *Job) []string { - var checks []string - var warning string - job.Seconds, warning = validateField(job.Seconds, numericPattern, "Seconds") - if warning != "" { - checks = append(checks, warning) - } - job.Minutes, warning = validateField(job.Minutes, numericPattern, "Minutes") - if warning != "" { - checks = append(checks, warning) - } - job.Hours, warning = validateField(job.Hours, hoursPattern, "Hours") - if warning != "" { - checks = append(checks, warning) - } - job.DayOfMonth, warning = validateField(job.DayOfMonth, dayOfMonthPattern, "DayOfMonth") - if warning != "" { - checks = append(checks, warning) - } - job.Month, warning = validateField(job.Month, monthPattern, "Month") - if warning != "" { - checks = append(checks, warning) - } - job.DayOfWeek, warning = validateField(job.DayOfWeek, dayOfWeekPattern, "DayOfWeek") - if warning != "" { - checks = append(checks, warning) - } - - return checks -} - -// Helper function for validation -func validateField(value, pattern, fieldName string) (string, string) { - match, _ := regexp.MatchString(pattern, value) - if !match { - return defaultValues[fieldName], fmt.Sprintf("invalid value for %s: %s, using default value %s", fieldName, value, defaultValues[fieldName]) - } - return value, "" -} diff --git a/internal/images/get-images.go b/internal/images/get-images.go deleted file mode 100644 index e172551..0000000 --- a/internal/images/get-images.go +++ /dev/null @@ -1,72 +0,0 @@ -package images - -import ( - "encoding/json" - "fmt" - "io" - "log" - "net/http" - "os" - "time" -) - -type ImageList struct { - RegistryURL string `json:"registryUrl"` - Repositories []struct { - Repository string `json:"repository"` - Images []struct { - Name string `json:"name"` - } `json:"images"` - } `json:"repositories"` -} - -type Image struct { - ID int `json:"ID"` - Registry string `json:"Registry"` - Repository string `json:"Repository"` - Tag string `json:"Tag"` - Digest string `json:"Digest"` - CreatedAt time.Time `json:"CreatedAt"` - UpdatedAt time.Time `json:"UpdatedAt"` -} - -func GetImages(url string) (string, error) { - token := os.Getenv("TOKEN") - if token == "" { - log.Fatal("Token cannot be empty") - } - - bearerToken := fmt.Sprintf("Bearer %s", token) - - req, err := http.NewRequest(http.MethodGet, url, nil) - if err != nil { - log.Printf("error in creating request: %v", err) - return "", err - } - req.Header.Add("Authorization", bearerToken) - - res, err := http.DefaultClient.Do(req) - if err != nil { - log.Println(err) - return "", err - } - defer res.Body.Close() - body, _ := io.ReadAll(res.Body) - - // snippet only - var result []Image - if err := json.Unmarshal(body, &result); err != nil { // Parse []byte to go struct pointer - log.Println("Can not unmarshal JSON") - return "", err - } - - fmt.Println(result) - - for _, img := range result { - url := fmt.Sprintf("http://%s/%s", img.Registry, img.Repository) - fmt.Println("url: ", url) - return url, nil - } - - return "", nil -} diff --git a/internal/satellite/satellite.go b/internal/satellite/satellite.go index 5f7cec3..5310a84 100644 --- a/internal/satellite/satellite.go +++ b/internal/satellite/satellite.go @@ -7,51 +7,68 @@ import ( "container-registry.com/harbor-satellite/internal/notifier" "container-registry.com/harbor-satellite/internal/scheduler" "container-registry.com/harbor-satellite/internal/state" - "container-registry.com/harbor-satellite/internal/utils" "container-registry.com/harbor-satellite/logger" ) +type RegistryConfig struct { + URL string + UserName string + Password string +} + +func NewRegistryConfig(url, username, password string) RegistryConfig { + return RegistryConfig{ + URL: url, + UserName: username, + Password: password, + } +} + type Satellite struct { - stateReader state.StateReader - schedulerKey scheduler.SchedulerKey + schedulerKey scheduler.SchedulerKey + LocalRegistryConfig RegistryConfig + SourcesRegistryConfig RegistryConfig + UseUnsecure bool + states []string } -func NewSatellite(ctx context.Context, schedulerKey scheduler.SchedulerKey) *Satellite { +func NewSatellite(ctx context.Context, schedulerKey scheduler.SchedulerKey, localRegistryConfig, sourceRegistryConfig RegistryConfig, useUnsecure bool, states []string) *Satellite { return &Satellite{ - schedulerKey: schedulerKey, + schedulerKey: schedulerKey, + LocalRegistryConfig: localRegistryConfig, + SourcesRegistryConfig: sourceRegistryConfig, + UseUnsecure: useUnsecure, + states: states, } } func (s *Satellite) Run(ctx context.Context) error { log := logger.FromContext(ctx) log.Info().Msg("Starting Satellite") - var fetchStateProcessCron string - state_fetch_period := config.GetStateFetchPeriod() - fetchConfigProcessCron, err := utils.FormatDuration(config.GetConfigFetchPeriod()) + replicateStateCron, err := config.GetJobSchedule(config.ReplicateStateJobName) if err != nil { - log.Warn().Msgf("Error formatting duration in seconds: %v", err) - log.Warn().Msgf("Using default duration: %v", state.DefaultFetchConfigFromGroundControlTimePeriod) - fetchConfigProcessCron = state.DefaultFetchConfigFromGroundControlTimePeriod + log.Error().Err(err).Msg("Error getting schedule") + return err } - fetchStateProcessCron, err = utils.FormatDuration(state_fetch_period) + updateConfigCron, err := config.GetJobSchedule(config.UpdateConfigJobName) if err != nil { - log.Warn().Msgf("Error formatting duration in seconds: %v", err) - log.Warn().Msgf("Using default duration: %v", state.DefaultFetchAndReplicateStateTimePeriod) - fetchStateProcessCron = state.DefaultFetchAndReplicateStateTimePeriod + log.Error().Err(err).Msg("Error getting schedule") + return err + } + ztrCron, err := config.GetJobSchedule(config.ZTRConfigJobName) + if err != nil { + log.Error().Err(err).Msg("Error getting schedule") + return err } - userName := config.GetHarborUsername() - password := config.GetHarborPassword() - zotURL := config.GetZotURL() - sourceRegistry := utils.FormatRegistryURL(config.GetRemoteRegistryURL()) - useUnsecure := config.UseUnsecure() // Get the scheduler from the context scheduler := ctx.Value(s.schedulerKey).(scheduler.Scheduler) // Create a simple notifier and add it to the process notifier := notifier.NewSimpleNotifier(ctx) // Creating a process to fetch and replicate the state states := config.GetStates() - fetchAndReplicateStateProcess := state.NewFetchAndReplicateStateProcess(fetchStateProcessCron, notifier, userName, password, zotURL, sourceRegistry, useUnsecure, states) - configFetchProcess := state.NewFetchConfigFromGroundControlProcess(fetchConfigProcessCron, "", "") + fetchAndReplicateStateProcess := state.NewFetchAndReplicateStateProcess(replicateStateCron, notifier, s.SourcesRegistryConfig.URL, s.SourcesRegistryConfig.UserName, s.SourcesRegistryConfig.Password, s.LocalRegistryConfig.URL, s.LocalRegistryConfig.UserName, s.LocalRegistryConfig.Password, s.UseUnsecure, states) + configFetchProcess := state.NewFetchConfigFromGroundControlProcess(updateConfigCron, "", "") + ztrProcess := state.NewZtrProcess(ztrCron) err = scheduler.Schedule(configFetchProcess) if err != nil { log.Error().Err(err).Msg("Error scheduling process") @@ -64,7 +81,6 @@ func (s *Satellite) Run(ctx context.Context) error { return err } // Schedule Register Satellite Process - ztrProcess := state.NewZtrProcess(state.DefaultZeroTouchRegistrationCronExpr) err = scheduler.Schedule(ztrProcess) if err != nil { log.Error().Err(err).Msg("Error scheduling process") diff --git a/internal/scheduler/event.go b/internal/scheduler/event.go index 0c02896..cdfa921 100644 --- a/internal/scheduler/event.go +++ b/internal/scheduler/event.go @@ -48,19 +48,22 @@ func (b *EventBroker) Subscribe(eventName string) <-chan Event { // Publish would take in the event and would emit the event to all the listeners. Would iterate over the subscribers array of the // event name and emit the event to all the listeners func (b *EventBroker) Publish(event Event, ctx context.Context) error { - b.mu.RLock() - defer b.mu.RUnlock() + b.mu.RLock() + defer b.mu.RUnlock() - for _, ch := range b.subscribers[event.Name] { - select { - case ch <- event: - case <-ctx.Done(): - return ctx.Err() - default: - return fmt.Errorf("event %s is not being consumed by the listeners", event.Name) - } - } - return nil + if subscribers, found := b.subscribers[event.Name]; found { + for _, ch := range subscribers { + select { + case ch <- event: + // Successfully sent the event. + case <-ctx.Done(): + return ctx.Err() + default: + fmt.Printf("Warning: Subscriber not consuming event %s\n", event.Name) + } + } + } + return nil } // Close cleans up all channels to prevent leaks. @@ -68,9 +71,30 @@ func (b *EventBroker) Close() { b.mu.Lock() defer b.mu.Unlock() - for _, channels := range b.subscribers { + for eventName, channels := range b.subscribers { for _, ch := range channels { - close(ch) + select { + case <-ch: + // Channel is already closed, skip closing it again. + default: + close(ch) + } } + delete(b.subscribers, eventName) } } + +func (b *EventBroker) Unsubscribe(eventName string, ch <-chan Event) { + b.mu.Lock() + defer b.mu.Unlock() + + if subscribers, found := b.subscribers[eventName]; found { + for i, subscriber := range subscribers { + if subscriber == ch { + b.subscribers[eventName] = append(subscribers[:i], subscribers[i+1:]...) + close(subscriber) + break + } + } + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 9ca0658..c443f5e 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -13,12 +13,17 @@ type SchedulerKey string const BasicSchedulerKey SchedulerKey = "basic-scheduler" const StopProcessEventName string = "stop-process-event" +const StopAllProcessesEventName string = "stop-all-processes-event" type StopProcessEventPayload struct { Id cron.EntryID ProcessName string } +type StopAllProcessesPayload struct { + Message string +} + type Scheduler interface { // GetSchedulerKey would return the key of the scheduler which is unique and for a particular scheduler // and is used to get the scheduler from the context @@ -105,6 +110,7 @@ func (s *BasicScheduler) Stop() error { s.mu.Lock() defer s.mu.Unlock() s.stopped = true + s.EventBroker.Close() s.cron.Stop() return nil } @@ -127,7 +133,12 @@ func (s *BasicScheduler) ListenForProcessEvent(ctx context.Context) { payload := event.Payload.(StopProcessEventPayload) log.Info().Msgf("Stopping process %s, with cron id %d", payload.ProcessName, payload.Id) s.StopProcess(payload.Id) + case event := <-s.EventBroker.Subscribe(StopAllProcessesEventName): + payload := event.Payload.(StopAllProcessesPayload) + log.Warn().Msgf("Cancelling all processes: %s", payload.Message) + s.EventBroker.Close() case <-ctx.Done(): + s.EventBroker.Close() log.Info().Msg("Scheduler is stopping listening for events ...") return } diff --git a/internal/state/config_process.go b/internal/state/config_process.go index 8bc8468..9e03a1f 100644 --- a/internal/state/config_process.go +++ b/internal/state/config_process.go @@ -5,15 +5,11 @@ import ( "fmt" "sync" + "container-registry.com/harbor-satellite/internal/config" "container-registry.com/harbor-satellite/internal/scheduler" - "container-registry.com/harbor-satellite/logger" "github.com/robfig/cron/v3" ) -const FetchConfigFromGroundControlProcessName string = "fetch-config-from-ground-control-process" - -const DefaultFetchConfigFromGroundControlTimePeriod string = "00h00m030s" - const FetchConfigFromGroundControlEventName string = "fetch-config-from-ground-control-event" const GroundControlSyncPath string = "/satellites/sync" @@ -31,7 +27,7 @@ type FetchConfigFromGroundControlProcess struct { func NewFetchConfigFromGroundControlProcess(cronExpr string, token string, groundControlURL string) *FetchConfigFromGroundControlProcess { return &FetchConfigFromGroundControlProcess{ - name: FetchConfigFromGroundControlProcessName, + name: config.UpdateConfigJobName, cronExpr: cronExpr, isRunning: false, token: token, @@ -56,21 +52,12 @@ func NewGroundControlConfigEvent(states []string) scheduler.Event { Payload: GroundControlPayload{ States: states, }, - Source: FetchConfigFromGroundControlProcessName, + Source: config.UpdateConfigJobName, } } func (f *FetchConfigFromGroundControlProcess) Execute(ctx context.Context) error { - log := logger.FromContext(ctx) - log.Info().Msgf("Starting process %s", f.name) - if !f.start() { - log.Warn().Msg("Process is already running") - return nil - } - defer f.stop() - log.Info().Msg("Fetching config from ground control") - event := NewGroundControlConfigEvent([]string{"state1", "state2"}) - f.eventBroker.Publish(event, ctx) + // TODO: Implement the logic to fetch the configuration from Ground Control one the endpoint is available on the Ground Control side return nil } @@ -87,7 +74,7 @@ func (f *FetchConfigFromGroundControlProcess) GetName() string { } func (f *FetchConfigFromGroundControlProcess) GetCronExpr() string { - return fmt.Sprintf("@every %s", f.cronExpr) + return f.cronExpr } func (f *FetchConfigFromGroundControlProcess) IsRunning() bool { @@ -95,12 +82,10 @@ func (f *FetchConfigFromGroundControlProcess) IsRunning() bool { } func (f *FetchConfigFromGroundControlProcess) CanExecute(ctx context.Context) (bool, string) { - return true, fmt.Sprintf("Process %s can execute all condition fulfilled", f.name) + return false, fmt.Sprintf("Process %s can execute all condition fulfilled", f.name) } func (f *FetchConfigFromGroundControlProcess) AddEventBroker(eventBroker *scheduler.EventBroker, ctx context.Context) { - log := logger.FromContext(ctx) - log.Info().Msgf("Adding event broker to process %s", f.name) f.eventBroker = eventBroker } diff --git a/internal/state/helpers.go b/internal/state/helpers.go index 331216d..848ef9d 100644 --- a/internal/state/helpers.go +++ b/internal/state/helpers.go @@ -36,7 +36,7 @@ func validateFilePath(path string, log *zerolog.Logger) error { func processURLInput(input, username, password string, log *zerolog.Logger) (StateFetcher, error) { log.Info().Msg("Input is a valid URL") - config.SetRemoteRegistryURL(input) + config.SetSourceRegistryURL(input) stateArtifactFetcher := NewURLStateFetcher(input, username, password) diff --git a/internal/state/registration_process.go b/internal/state/registration_process.go index 7b4847b..1117947 100644 --- a/internal/state/registration_process.go +++ b/internal/state/registration_process.go @@ -10,13 +10,12 @@ import ( "container-registry.com/harbor-satellite/internal/config" "container-registry.com/harbor-satellite/internal/scheduler" + "container-registry.com/harbor-satellite/internal/utils" "container-registry.com/harbor-satellite/logger" "github.com/robfig/cron/v3" ) const ZeroTouchRegistrationRoute = "satellites/ztr" -const ZeroTouchRegistrationProcessName = "zero-touch-registration-process" -const DefaultZeroTouchRegistrationCronExpr string = "00h00m05s" const ZeroTouchRegistrationEventName = "zero-touch-registration-event" type ZtrProcess struct { @@ -36,7 +35,7 @@ type ZtrProcess struct { func NewZtrProcess(cronExpr string) *ZtrProcess { return &ZtrProcess{ - Name: ZeroTouchRegistrationProcessName, + Name: config.ZTRConfigJobName, cronExpr: cronExpr, mu: &sync.Mutex{}, } @@ -66,18 +65,18 @@ func (z *ZtrProcess) Execute(ctx context.Context) error { log.Error().Msgf("Failed to register satellite: %v", err) return err } - if stateConfig.Auth.Name == "" || stateConfig.Auth.Secret == "" || stateConfig.Auth.Registry == "" { - log.Error().Msgf("Failed to register satellite: %v", err) - return fmt.Errorf("failed to register satellite: %w", err) + if stateConfig.Auth.SourceUsername == "" || stateConfig.Auth.SourcePassword == "" || stateConfig.Auth.Registry == "" { + log.Error().Msgf("Failed to register satellite: invalid state auth config received") + return fmt.Errorf("failed to register satellite: invalid state auth config received") } // Update the state config in app config - config.UpdateStateConfig(stateConfig.Auth.Name, stateConfig.Auth.Registry, stateConfig.Auth.Secret, stateConfig.States) + config.UpdateStateAuthConfig(stateConfig.Auth.SourceUsername, stateConfig.Auth.Registry, stateConfig.Auth.SourcePassword, stateConfig.States) zeroTouchRegistrationEvent := scheduler.Event{ Name: ZeroTouchRegistrationEventName, Payload: ZeroTouchRegistrationEventPayload{ StateConfig: stateConfig, }, - Source: ZeroTouchRegistrationProcessName, + Source: z.Name, } z.eventBroker.Publish(zeroTouchRegistrationEvent, ctx) stopProcessPayload := scheduler.StopProcessEventPayload{ @@ -87,7 +86,7 @@ func (z *ZtrProcess) Execute(ctx context.Context) error { stopProcessEvent := scheduler.Event{ Name: scheduler.StopProcessEventName, Payload: stopProcessPayload, - Source: ZeroTouchRegistrationProcessName, + Source: z.Name, } z.eventBroker.Publish(stopProcessEvent, ctx) return nil @@ -106,7 +105,7 @@ func (z *ZtrProcess) GetName() string { } func (z *ZtrProcess) GetCronExpr() string { - return fmt.Sprintf("@every %s", z.cronExpr) + return z.cronExpr } func (z *ZtrProcess) IsRunning() bool { @@ -119,16 +118,10 @@ func (z *ZtrProcess) CanExecute(ctx context.Context) (bool, string) { log := logger.FromContext(ctx) log.Info().Msgf("Checking if process %s can execute", z.Name) errors, warnings := z.loadConfig() - if len(errors) > 0 || len(warnings) > 0 { - for _, warning := range warnings { - log.Warn().Msgf("Warning loading config: %v", warning) - } - for _, err := range errors { - log.Error().Msgf("Error loading config: %v", err) - } - return false, "error loading config" + err := utils.HandleErrorAndWarning(log, errors, warnings) + if err != nil { + return false, err.Error() } - checks := []struct { condition bool message string @@ -171,7 +164,7 @@ func (z *ZtrProcess) stop() { // loadConfig loads the configuration. // It returns an error if the configuration cannot be loaded. -func (z *ZtrProcess) loadConfig() ([]error, []string) { +func (z *ZtrProcess) loadConfig() ([]error, []config.Warning) { return config.InitConfig(config.DefaultConfigPath) } diff --git a/internal/state/replicator.go b/internal/state/replicator.go index cb72e40..4a74343 100644 --- a/internal/state/replicator.go +++ b/internal/state/replicator.go @@ -19,20 +19,24 @@ type Replicator interface { } type BasicReplicator struct { - username string - password string useUnsecure bool - remoteRegistryURL string + sourceUsername string + sourcePassword string sourceRegistry string + remoteRegistryURL string + remoteUsername string + remotePassword string } -func NewBasicReplicator(username, password, zotURL, sourceRegistry string, useUnsecure bool) Replicator { +func NewBasicReplicator(sourceUsername, sourcePassword, sourceRegistry, remoteURL, remoteUsername, remotePassword string, useUnsecure bool) Replicator { return &BasicReplicator{ - username: username, - password: password, + sourceUsername: sourceUsername, + sourcePassword: sourcePassword, useUnsecure: useUnsecure, - remoteRegistryURL: zotURL, + remoteRegistryURL: remoteURL, sourceRegistry: sourceRegistry, + remoteUsername: remoteUsername, + remotePassword: remotePassword, } } @@ -59,22 +63,28 @@ func (e Entity) GetTag() string { // Replicate replicates images from the source registry to the Zot registry. func (r *BasicReplicator) Replicate(ctx context.Context, replicationEntities []Entity) error { log := logger.FromContext(ctx) - auth := authn.FromConfig(authn.AuthConfig{ - Username: r.username, - Password: r.password, + pullAuthConfig := authn.FromConfig(authn.AuthConfig{ + Username: r.sourceUsername, + Password: r.sourcePassword, + }) + pushAuthConfig := authn.FromConfig(authn.AuthConfig{ + Username: r.remoteUsername, + Password: r.remotePassword, }) - options := []crane.Option{crane.WithAuth(auth)} + pullOptions := []crane.Option{crane.WithAuth(pullAuthConfig)} + pushOptions := []crane.Option{crane.WithAuth(pushAuthConfig)} + if r.useUnsecure { - options = append(options, crane.Insecure) + pullOptions = append(pullOptions, crane.Insecure) + pushOptions = append(pushOptions, crane.Insecure) } for _, replicationEntity := range replicationEntities { log.Info().Msgf("Pulling image %s from repository %s at registry %s with tag %s", replicationEntity.GetName(), replicationEntity.GetRepository(), r.sourceRegistry, replicationEntity.GetTag()) - // Pull the image from the source registry - srcImage, err := crane.Pull(fmt.Sprintf("%s/%s/%s:%s", r.sourceRegistry, replicationEntity.GetRepository(), replicationEntity.GetName(), replicationEntity.GetTag()), options...) + srcImage, err := crane.Pull(fmt.Sprintf("%s/%s/%s:%s", r.sourceRegistry, replicationEntity.GetRepository(), replicationEntity.GetName(), replicationEntity.GetTag()), pullOptions...) if err != nil { log.Error().Msgf("Failed to pull image: %v", err) return err @@ -84,7 +94,7 @@ func (r *BasicReplicator) Replicate(ctx context.Context, replicationEntities []E ociImage := mutate.MediaType(srcImage, types.OCIManifestSchema1) // Push the converted OCI image to the Zot registry - err = crane.Push(ociImage, fmt.Sprintf("%s/%s/%s:%s", r.remoteRegistryURL, replicationEntity.GetRepository(), replicationEntity.GetName(), replicationEntity.GetTag()), options...) + err = crane.Push(ociImage, fmt.Sprintf("%s/%s/%s:%s", r.remoteRegistryURL, replicationEntity.GetRepository(), replicationEntity.GetName(), replicationEntity.GetTag()), pushOptions...) if err != nil { log.Error().Msgf("Failed to push image: %v", err) return err @@ -98,8 +108,8 @@ func (r *BasicReplicator) Replicate(ctx context.Context, replicationEntities []E func (r *BasicReplicator) DeleteReplicationEntity(ctx context.Context, replicationEntity []Entity) error { log := logger.FromContext(ctx) auth := authn.FromConfig(authn.AuthConfig{ - Username: r.username, - Password: r.password, + Username: r.remoteUsername, + Password: r.remotePassword, }) options := []crane.Option{crane.WithAuth(auth)} diff --git a/internal/state/state_process.go b/internal/state/state_process.go index af6f652..68c6bd6 100644 --- a/internal/state/state_process.go +++ b/internal/state/state_process.go @@ -6,6 +6,7 @@ import ( "strings" "sync" + "container-registry.com/harbor-satellite/internal/config" "container-registry.com/harbor-satellite/internal/notifier" "container-registry.com/harbor-satellite/internal/scheduler" "container-registry.com/harbor-satellite/internal/utils" @@ -14,16 +15,14 @@ import ( "github.com/rs/zerolog" ) -const FetchAndReplicateStateProcessName string = "fetch-replicate-state-process" - -const DefaultFetchAndReplicateStateTimePeriod string = "00h00m010s" - type FetchAndReplicateAuthConfig struct { - Username string - Password string - UseUnsecure bool - RemoteRegistryURL string - SourceRegistry string + SourceRegistry string + SourceRegistryUserName string + SourceRegistryPassword string + UseUnsecure bool + RemoteRegistryURL string + RemoteRegistryUserName string + RemoteRegistryPassword string } type FetchAndReplicateStateProcess struct { @@ -36,6 +35,7 @@ type FetchAndReplicateStateProcess struct { mu *sync.Mutex authConfig FetchAndReplicateAuthConfig eventBroker *scheduler.EventBroker + Replicator Replicator } type StateMap struct { @@ -52,21 +52,26 @@ func NewStateMap(url []string) []StateMap { return stateMap } -func NewFetchAndReplicateStateProcess(cronExpr string, notifier notifier.Notifier, username, password, remoteRegistryURL, sourceRegistryURL string, useUnsecure bool, states []string) *FetchAndReplicateStateProcess { +func NewFetchAndReplicateStateProcess(cronExpr string, notifier notifier.Notifier, sourceRegistryURL, sourceRegistryUsername, sourceRegistryPassword, remoteRegistryURL, remoteRegistryUsername, remoteRegistryPassword string, useUnsecure bool, states []string) *FetchAndReplicateStateProcess { + sourceURL := utils.FormatRegistryURL(sourceRegistryURL) + remoteURL := utils.FormatRegistryURL(remoteRegistryURL) return &FetchAndReplicateStateProcess{ - name: FetchAndReplicateStateProcessName, + name: config.ReplicateStateJobName, cronExpr: cronExpr, isRunning: false, notifier: notifier, mu: &sync.Mutex{}, stateMap: NewStateMap(states), authConfig: FetchAndReplicateAuthConfig{ - Username: username, - Password: password, - UseUnsecure: useUnsecure, - RemoteRegistryURL: remoteRegistryURL, - SourceRegistry: sourceRegistryURL, + SourceRegistry: sourceURL, + SourceRegistryUserName: sourceRegistryUsername, + SourceRegistryPassword: sourceRegistryPassword, + UseUnsecure: useUnsecure, + RemoteRegistryURL: remoteURL, + RemoteRegistryUserName: remoteRegistryUsername, + RemoteRegistryPassword: remoteRegistryPassword, }, + Replicator: NewBasicReplicator(sourceRegistryUsername, sourceRegistryPassword, sourceURL, remoteURL, remoteRegistryUsername, remoteRegistryPassword, useUnsecure), } } @@ -77,8 +82,8 @@ func (f *FetchAndReplicateStateProcess) Execute(ctx context.Context) error { log.Warn().Msgf("Process %s is already running", f.name) return nil } - bool, reason := f.CanExecute(ctx) - if !bool { + canExecute, reason := f.CanExecute(ctx) + if !canExecute { log.Warn().Msgf("Cannot execute process: %s", reason) return nil } @@ -86,7 +91,7 @@ func (f *FetchAndReplicateStateProcess) Execute(ctx context.Context) error { for i := range f.stateMap { log.Info().Msgf("Processing state for %s", f.stateMap[i].url) - stateFetcher, err := processInput(f.stateMap[i].url, f.authConfig.Username, f.authConfig.Password, log) + stateFetcher, err := processInput(f.stateMap[i].url, f.authConfig.SourceRegistryUserName, f.authConfig.SourceRegistryPassword, log) if err != nil { log.Error().Err(err).Msg("Error processing input") return err @@ -102,15 +107,13 @@ func (f *FetchAndReplicateStateProcess) Execute(ctx context.Context) error { if err := f.notifier.Notify(); err != nil { log.Error().Err(err).Msg("Error sending notification") } - - replicator := NewBasicReplicator(f.authConfig.Username, f.authConfig.Password, f.authConfig.RemoteRegistryURL, f.authConfig.SourceRegistry, f.authConfig.UseUnsecure) // Delete the entities from the remote registry - if err := replicator.DeleteReplicationEntity(ctx, deleteEntity); err != nil { + if err := f.Replicator.DeleteReplicationEntity(ctx, deleteEntity); err != nil { log.Error().Err(err).Msg("Error deleting entities") return err } // Replicate the entities to the remote registry - if err := replicator.Replicate(ctx, replicateEntity); err != nil { + if err := f.Replicator.Replicate(ctx, replicateEntity); err != nil { log.Error().Err(err).Msg("Error replicating state") return err } @@ -179,7 +182,7 @@ func (f *FetchAndReplicateStateProcess) GetName() string { } func (f *FetchAndReplicateStateProcess) GetCronExpr() string { - return fmt.Sprintf("@every %s", f.cronExpr) + return f.cronExpr } func (f *FetchAndReplicateStateProcess) IsRunning() bool { @@ -194,8 +197,8 @@ func (f *FetchAndReplicateStateProcess) CanExecute(ctx context.Context) (bool, s {f.stateMap == nil, "state map is nil"}, {f.authConfig.RemoteRegistryURL == "", "remote registry URL is empty"}, {f.authConfig.SourceRegistry == "", "source registry is empty"}, - {f.authConfig.Username == "", "username is empty"}, - {f.authConfig.Password == "", "password is empty"}, + {f.authConfig.SourceRegistryUserName == "", "username is empty"}, + {f.authConfig.SourceRegistryPassword == "", "password is empty"}, } var missingFields []string @@ -290,34 +293,41 @@ func (f *FetchAndReplicateStateProcess) AddEventBroker(eventBroker *scheduler.Ev func (f *FetchAndReplicateStateProcess) ListenForUpdatedConfig(ctx context.Context) { log := logger.FromContext(ctx) - log.Info().Msg("Listening for updated config from ground control") + log.Info().Msgf("Process %s is listening for updated config", f.name) + fetchConfigCh := f.eventBroker.Subscribe(FetchConfigFromGroundControlEventName) + zeroTouchRegistrationCh := f.eventBroker.Subscribe(ZeroTouchRegistrationEventName) + + defer func() { + log.Info().Msgf("Process %s unsubscribing from %s and %s", f.name, FetchConfigFromGroundControlEventName, ZeroTouchRegistrationEventName) + f.eventBroker.Unsubscribe(FetchConfigFromGroundControlEventName, fetchConfigCh) + f.eventBroker.Unsubscribe(ZeroTouchRegistrationEventName, zeroTouchRegistrationCh) + }() + for { select { case <-ctx.Done(): return - case event := <-f.eventBroker.Subscribe(FetchConfigFromGroundControlEventName): - log.Info().Msg("Received updated config from ground control") - payload, ok := event.Payload.(GroundControlPayload) - if !ok { - log.Error().Msgf("received invalid payload from %s, for process %s", event.Source, FetchConfigFromGroundControlEventName) - return - } - log.Info().Msgf("Received updated config from ground control with states: %v", len(payload.States)) - case event := <-f.eventBroker.Subscribe(ZeroTouchRegistrationEventName): - log.Info().Msgf("Received %s event with source %s", event.Name, event.Source) - payload, ok := event.Payload.(ZeroTouchRegistrationEventPayload) - if !ok { - log.Error().Msgf("Received invalid payload from %s, for process %s", event.Source, ZeroTouchRegistrationEventName) - return - } - f.UpdateFetchProcessConfigFromZtr(payload.StateConfig.Auth.Name, payload.StateConfig.Auth.Secret, payload.StateConfig.Auth.Registry, payload.StateConfig.States) + case event := <-fetchConfigCh: + log.Info().Msgf("Received updated config from ground control from source %s", event.Source) + case event := <-zeroTouchRegistrationCh: + f.HandelPayloadFromZTR(event, log) } } } +func (f *FetchAndReplicateStateProcess) HandelPayloadFromZTR(event scheduler.Event, log *zerolog.Logger) { + log.Info().Msgf("Received %s event with source %s", event.Name, event.Source) + payload, ok := event.Payload.(ZeroTouchRegistrationEventPayload) + if !ok { + log.Error().Msgf("Received invalid payload from %s, for process %s", event.Source, ZeroTouchRegistrationEventName) + return + } + f.UpdateFetchProcessConfigFromZtr(payload.StateConfig.Auth.SourceUsername, payload.StateConfig.Auth.SourcePassword, payload.StateConfig.Auth.Registry, payload.StateConfig.States) +} + func (f *FetchAndReplicateStateProcess) UpdateFetchProcessConfigFromZtr(username, password, sourceRegistryURL string, states []string) { - f.authConfig.Username = username - f.authConfig.Password = password + f.authConfig.SourceRegistryUserName = username + f.authConfig.SourceRegistryPassword = password f.authConfig.SourceRegistry = utils.FormatRegistryURL(sourceRegistryURL) // The states contain all the states that this satellite needs to track thus we would have to add the new states to the state map diff --git a/internal/store/file-fetch.go b/internal/store/file-fetch.go deleted file mode 100644 index fe32f63..0000000 --- a/internal/store/file-fetch.go +++ /dev/null @@ -1,82 +0,0 @@ -package store - -import ( - "context" - "encoding/json" - "os" - "path/filepath" - - "container-registry.com/harbor-satellite/logger" -) - -type FileImageList struct { - Path string -} - -type Repository struct { - Repository string `json:"repository"` - Images []struct { - Name string `json:"name"` - } `json:"images"` -} - -type ImageData struct { - RegistryUrl string `json:"registryUrl"` - Repositories []Repository `json:"repositories"` -} - -func (f *FileImageList) Type(ctx context.Context) string { - return "File" -} - -func FileImageListFetcher(ctx context.Context, relativePath string) *FileImageList { - log := logger.FromContext(ctx) - // Get the current working directory - dir, err := os.Getwd() - if err != nil { - log.Error().Err(err).Msg("Error getting current directory") - return nil - } - - // Construct the absolute path from the relative path - absPath := filepath.Join(dir, relativePath) - - return &FileImageList{ - Path: absPath, - } -} - -func (client *FileImageList) List(ctx context.Context) ([]Image, error) { - log := logger.FromContext(ctx) - var images []Image - - // Read the file - data, err := os.ReadFile(client.Path) - if err != nil { - log.Error().Err(err).Msg("Error reading file") - return nil, err - } - - var imageData ImageData - // Parse the JSON data - err = json.Unmarshal(data, &imageData) - if err != nil { - log.Error().Err(err).Msg("Error unmarshalling JSON data") - return nil, err - } - - // Iterate over the repositories - for _, repo := range imageData.Repositories { - // Iterate over the images in each repository - for _, image := range repo.Images { - // Add each "name" value to the images slice - images = append(images, Image{Name: image.Name}) - } - } - - return images, nil -} - -func (client *FileImageList) GetDigest(ctx context.Context, tag string) (string, error) { - return "Not implemented yet", nil -} diff --git a/internal/store/http-fetch.go b/internal/store/http-fetch.go deleted file mode 100644 index 4dc7b7c..0000000 --- a/internal/store/http-fetch.go +++ /dev/null @@ -1,133 +0,0 @@ -package store - -import ( - "context" - "encoding/base64" - "encoding/json" - "fmt" - "io" - "net/http" - "os" - "strings" - "time" - - "container-registry.com/harbor-satellite/internal/config" - "container-registry.com/harbor-satellite/logger" - "github.com/google/go-containerregistry/pkg/authn" - "github.com/google/go-containerregistry/pkg/crane" -) - -type RemoteImageList struct { - BaseURL string - username string - password string - use_unsecure bool - zot_url string -} - -type TagListResponse struct { - Name string `json:"name"` - Tags []string `json:"tags"` -} - -func RemoteImageListFetcher(ctx context.Context, url string) *RemoteImageList { - return &RemoteImageList{ - BaseURL: url, - username: config.GetHarborUsername(), - password: config.GetHarborPassword(), - use_unsecure: config.UseUnsecure(), - zot_url: config.GetZotURL(), - } -} - -func (r *RemoteImageList) Type(ctx context.Context) string { - return "Remote" -} - -func (client *RemoteImageList) List(ctx context.Context) ([]Image, error) { - log := logger.FromContext(ctx) - // Construct the URL for fetching tags - url := client.BaseURL + "/tags/list" - - // Encode credentials for Basic Authentication - username := os.Getenv("HARBOR_USERNAME") - password := os.Getenv("HARBOR_PASSWORD") - auth := base64.StdEncoding.EncodeToString([]byte(username + ":" + password)) - - // Create a new HTTP request - req, err := http.NewRequest("GET", url, nil) - if err != nil { - log.Error().Msgf("failed to create request: %v", err) - return nil, err - } - - // Set the Authorization header - req.Header.Set("Authorization", "Basic "+auth) - - // Configure the HTTP client with a timeout - httpClient := &http.Client{ - Timeout: 5 * time.Second, - } - - // Send the HTTP request - log.Info().Msgf("Sending request to %s", url) - resp, err := httpClient.Do(req) - if err != nil { - log.Error().Msgf("failed to send request: %v", err) - return nil, err - } - defer resp.Body.Close() - - // Read the response body - body, err := io.ReadAll(resp.Body) - if err != nil { - log.Error().Msgf("failed to read response body: %v", err) - return nil, err - } - - // Unmarshal the JSON response - var tagListResponse TagListResponse - if err := json.Unmarshal(body, &tagListResponse); err != nil { - log.Error().Msgf("failed to unmarshal response: %v", err) - return nil, err - } - - // Prepare a slice to store the images - var images []Image - - // Iterate over the tags and construct the image references - for _, tag := range tagListResponse.Tags { - images = append(images, Image{ - Name: fmt.Sprintf("%s:%s", tagListResponse.Name, tag), - }) - } - return images, nil -} - -func (client *RemoteImageList) GetDigest(ctx context.Context, tag string) (string, error) { - log := logger.FromContext(ctx) - // Construct the image reference - imageRef := fmt.Sprintf("%s:%s", client.BaseURL, tag) - // Remove extra characters from the URL - imageRef = imageRef[strings.Index(imageRef, "//")+2:] - imageRef = strings.ReplaceAll(imageRef, "/v2", "") - - // Encode credentials for Basic Authentication - username := os.Getenv("HARBOR_USERNAME") - password := os.Getenv("HARBOR_PASSWORD") - auth := &authn.Basic{Username: username, Password: password} - // Prepare options for crane.Digest - options := []crane.Option{crane.WithAuth(auth)} - if client.use_unsecure { - options = append(options, crane.Insecure) - } - - // Use crane.Digest to get the digest of the image - digest, err := crane.Digest(imageRef, options...) - if err != nil { - log.Error().Msgf("failed to get digest using crane: %v", err) - return "", err - } - - return digest, nil -} diff --git a/internal/store/in-memory-store.go b/internal/store/in-memory-store.go deleted file mode 100644 index da25d3f..0000000 --- a/internal/store/in-memory-store.go +++ /dev/null @@ -1,269 +0,0 @@ -package store - -import ( - "context" - "fmt" - "os" - "strings" - - "container-registry.com/harbor-satellite/logger" - "github.com/google/go-containerregistry/pkg/crane" -) - -type Image struct { - Digest string - Name string -} - -type inMemoryStore struct { - images map[string]string - fetcher ImageFetcher -} - -type Storer interface { - List(ctx context.Context) ([]Image, error) - Add(ctx context.Context, digest string, image string) error - Remove(ctx context.Context, digest string, image string) error -} - -type ImageFetcher interface { - List(ctx context.Context) ([]Image, error) - GetDigest(ctx context.Context, tag string) (string, error) - Type(ctx context.Context) string -} - -func NewInMemoryStore(ctx context.Context, fetcher ImageFetcher) (context.Context, Storer) { - return ctx, &inMemoryStore{ - images: make(map[string]string), - fetcher: fetcher, - } -} - -func (s *inMemoryStore) List(ctx context.Context) ([]Image, error) { - log := logger.FromContext(ctx) - var imageList []Image - var change bool - err := error(nil) - - // Fetch images from the file/remote source - imageList, err = s.fetcher.List(ctx) - if err != nil { - return nil, err - } - - // Handle File and Remote fetcher types differently - switch s.fetcher.Type(ctx) { - case "File": - for _, img := range imageList { - // Check if the image already exists in the store - if _, exists := s.images[img.Name]; !exists { - // Add the image to the store - s.AddImage(ctx, img.Name) - change = true - } else { - log.Info().Msgf("Image %s already exists in the store", img.Name) - } - } - - // Iterate over s.images and remove any image that is not found in imageList - for image := range s.images { - found := false - for _, img := range imageList { - if img.Name == image { - found = true - break - } - } - if !found { - s.RemoveImage(ctx, image) - change = true - } - } - - // Empty and refill imageList with the contents from s.images - imageList = imageList[:0] - for name, digest := range s.images { - imageList = append(imageList, Image{Name: name, Digest: digest}) - } - - // Print out the entire store for debugging purposes - log.Info().Msg("Current store:") - for image := range s.images { - log.Info().Msgf("Image: %s", image) - } - - case "Remote": - // Trim the imageList elements to remove the project name from the image reference - for i, img := range imageList { - parts := strings.Split(img.Name, "/") - if len(parts) > 1 { - // Take the second part as the new Reference - imageList[i].Name = parts[1] - } - } - // iterate over imageList and call GetDigest for each tag - for _, img := range imageList { - // Split the image reference to get the tag - tagParts := strings.Split(img.Name, ":") - // Check if there is a tag part, min length is 1 char - if len(tagParts) < 2 { - log.Error().Msgf("Invalid image reference: %s", img.Name) - } - // Use the last part as the tag - tag := tagParts[len(tagParts)-1] - // Get the digest for the tag - digest, err := s.fetcher.GetDigest(ctx, tag) - if err != nil { - return nil, err - } - - // Check if the image exists and matches the digest - if !(s.checkImageAndDigest(ctx, digest, img.Name)) { - change = true - } - - } - - // Create imageMap filled with all images from remote imageList - imageMap := make(map[string]bool) - for _, img := range imageList { - imageMap[img.Name] = true - } - - // Iterate over in memory store and remove any image that is not found in imageMap - for digest, image := range s.images { - if _, exists := imageMap[image]; !exists { - s.Remove(ctx, digest, image) - change = true - } - } - // Print out the entire store for debugging purposes - log.Info().Msg("Current store:") - for digest, imageRef := range s.images { - log.Info().Msgf("Image: %s, Digest: %s", imageRef, digest) - } - - // Empty and refill imageList with the contents from s.images - imageList = imageList[:0] - for _, name := range s.images { - imageList = append(imageList, Image{Digest: "", Name: name}) - } - - } - if change { - log.Info().Msg("Changes detected in the store") - change = false - return imageList, nil - } else { - log.Info().Msg("No changes detected in the store") - return nil, nil - } -} - -func (s *inMemoryStore) Add(ctx context.Context, digest string, image string) error { - log := logger.FromContext(ctx) - // Check if the image already exists in the store - if _, exists := s.images[digest]; exists { - log.Info().Msgf("Image: %s, digest: %s already exists in the store.", image, digest) - return fmt.Errorf("image %s already exists in the store", image) - } else { - // Add the image and its digest to the store - s.images[digest] = image - log.Info().Msgf("Successfully added image: %s, digest: %s", image, digest) - return nil - } -} - -func (s *inMemoryStore) AddImage(ctx context.Context, image string) error { - log := logger.FromContext(ctx) - // Add the image to the store - s.images[image] = "" - log.Info().Msgf("Added image: %s", image) - return nil -} - -func (s *inMemoryStore) Remove(ctx context.Context, digest string, image string) error { - log := logger.FromContext(ctx) - // Check if the image exists in the store - if _, exists := s.images[digest]; exists { - // Remove the image and its digest from the store - delete(s.images, digest) - log.Info().Msgf("Successfully removed image: %s, digest: %s", image, digest) - return nil - } else { - log.Warn().Msgf("Failed to remove image: %s, digest: %s. Not found in the store.", image, digest) - return fmt.Errorf("image %s not found in the store", image) - } -} - -func (s *inMemoryStore) RemoveImage(ctx context.Context, image string) error { - log := logger.FromContext(ctx) - // Remove the image from the store - delete(s.images, image) - log.Info().Msgf("Removed image: %s", image) - return nil -} - -// TODO: Rework complicated logic and add support for multiple repositories -// checkImageAndDigest checks if the image exists in the store and if the digest matches the image reference -func (s *inMemoryStore) checkImageAndDigest(ctx context.Context, digest string, image string) bool { - log := logger.FromContext(ctx) - - // Check if the received image exists in the store - for storeDigest, storeImage := range s.images { - if storeImage == image { - // Image exists, now check if the digest matches - if storeDigest == digest { - // Digest exists and matches the current image's - // Remove what comes before the ":" in image and set it to tag variable - tag := strings.Split(image, ":")[1] - localRegistryDigest, err := GetLocalDigest(context.Background(), tag) - if err != nil { - log.Error().Msgf("Error getting digest from local registry: %v", err) - return false - } else { - // Check if the digest from the local registry matches the digest from the store - if digest == localRegistryDigest { - return true - } else { - return false - } - } - } else { - // Digest exists but does not match the current image reference - s.Remove(context.Background(), storeDigest, storeImage) - s.Add(context.Background(), digest, image) - return false - } - } - } - - // Try to add the image to the store - // Add will check if it already exists in the store before adding - // If adding was successful, return true, else return false - err := s.Add(context.Background(), digest, image) - return err != nil -} - -func GetLocalDigest(ctx context.Context, tag string) (string, error) { - log := logger.FromContext(ctx) - - zotUrl := os.Getenv("ZOT_URL") - userURL := os.Getenv("USER_INPUT") - - // Remove extra characters from the URLs - userURL = userURL[strings.Index(userURL, "//")+2:] - userURL = strings.ReplaceAll(userURL, "/v2", "") - - // Construct the URL for fetching the digest - url := zotUrl + "/" + userURL + ":" + tag - - // Use crane.Digest to get the digest of the image - digest, err := crane.Digest(url) - if err != nil { - log.Error().Msgf("Error getting digest using crane: %v", err) - return "", fmt.Errorf("failed to get digest using crane: %w", err) - } - - return digest, nil -} diff --git a/internal/utils/utils.go b/internal/utils/utils.go index 64aff3e..0ff9453 100644 --- a/internal/utils/utils.go +++ b/internal/utils/utils.go @@ -16,6 +16,7 @@ import ( "container-registry.com/harbor-satellite/internal/config" "container-registry.com/harbor-satellite/logger" "container-registry.com/harbor-satellite/registry" + "github.com/rs/zerolog" "github.com/spf13/cobra" ) @@ -39,18 +40,22 @@ func ValidateRegistryAddress(registryAdr, registryPort string) (string, error) { // / HandleOwnRegistry handles the own registry address and port and sets the Zot URL func HandleOwnRegistry() error { - zotURL, err := ValidateRegistryAddress(config.GetOwnRegistryAdr(), config.GetOwnRegistryPort()) + _, err := url.Parse(config.GetRemoteRegistryURL()) if err != nil { - return err + return fmt.Errorf("error parsing URL: %w", err) } - config.SetZotURL(zotURL) + config.SetRemoteRegistryURL(FormatRegistryURL(config.GetRemoteRegistryURL())) return nil } // LaunchDefaultZotRegistry launches the default Zot registry using the Zot config path func LaunchDefaultZotRegistry() error { - defaultZotURL := fmt.Sprintf("%s:%s", "127.0.0.1", "8585") - config.SetZotURL(defaultZotURL) + defaultZotConfig, err := registry.ReadConfig(config.GetZotConfigPath()) + if err != nil { + return fmt.Errorf("error reading config: %w", err) + } + defaultZotURL := defaultZotConfig.GetLocalRegistryURL() + config.SetRemoteRegistryURL(defaultZotURL) launch, err := registry.LaunchRegistry(config.GetZotConfigPath()) if !launch { return fmt.Errorf("error launching registry: %w", err) @@ -95,34 +100,6 @@ func GetRepositoryAndImageNameFromArtifact(repository string) (string, string, e return repo, image, nil } -func FormatDuration(input string) (string, error) { - seconds, err := strconv.Atoi(input) // Convert input string to an integer - if err != nil { - return "", errors.New("invalid input: not a valid number") - } - if seconds < 0 { - return "", errors.New("invalid input: seconds cannot be negative") - } - - hours := seconds / 3600 - minutes := (seconds % 3600) / 60 - secondsRemaining := seconds % 60 - - var result string - - if hours > 0 { - result += strconv.Itoa(hours) + "h" - } - if minutes > 0 { - result += strconv.Itoa(minutes) + "m" - } - if secondsRemaining > 0 || result == "" { - result += strconv.Itoa(secondsRemaining) + "s" - } - - return result, nil -} - func SetupContext(context context.Context) (context.Context, context.CancelFunc) { ctx, cancel := signal.NotifyContext(context, syscall.SIGTERM, syscall.SIGINT) return ctx, cancel @@ -175,3 +152,27 @@ func WriteFile(path string, data []byte) error { } return nil } + +func HandleErrorAndWarning(log *zerolog.Logger, errors []error, warnings []config.Warning) error { + for i := range warnings { + log.Warn().Msg(string(warnings[i])) + } + for i := range errors { + log.Error().Msg(errors[i].Error()) + } + if len(errors) > 0 { + return fmt.Errorf("error initializing config") + } + return nil +} + +func CommandRunSetup(cmd *cobra.Command) error { + errors, warnings := config.InitConfig(config.DefaultConfigPath) + SetupContextForCommand(cmd) + log := logger.FromContext(cmd.Context()) + err := HandleErrorAndWarning(log, errors, warnings) + if err != nil { + return err + } + return nil +} diff --git a/registry/default_config.go b/registry/default_config.go index ec5bc2e..dc25f82 100644 --- a/registry/default_config.go +++ b/registry/default_config.go @@ -20,6 +20,7 @@ type DefaultZotConfig struct { Log struct { Level string `json:"level"` } `json:"log"` + RemoteURL string } func (c *DefaultZotConfig) GetLocalRegistryURL() string { @@ -50,6 +51,9 @@ func ReadConfig(filePath string) (*DefaultZotConfig, error) { if err != nil { return nil, fmt.Errorf("could not unmarshal JSON: %w", err) } - return &config, nil } + +func (c *DefaultZotConfig) SetZotRemoteURL(url string) { + c.RemoteURL = url +} diff --git a/value/io.containerd.metadata.v1.bolt/meta.db b/value/io.containerd.metadata.v1.bolt/meta.db deleted file mode 100644 index 1884558..0000000 Binary files a/value/io.containerd.metadata.v1.bolt/meta.db and /dev/null differ