diff --git a/internal/api/api.go b/internal/api/api.go index 9cb5656..e82fda1 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -12,13 +12,15 @@ import ( type ApiManagerCtx struct { logger zerolog.Logger rooms types.RoomManager + pull types.PullManager conf *config.API } -func New(roomManager types.RoomManager, conf *config.API) *ApiManagerCtx { +func New(rooms types.RoomManager, pull types.PullManager, conf *config.API) *ApiManagerCtx { return &ApiManagerCtx{ logger: log.With().Str("module", "api").Logger(), - rooms: roomManager, + rooms: rooms, + pull: pull, conf: conf, } } diff --git a/internal/api/pull.go b/internal/api/pull.go index 407fe0a..7787a1a 100644 --- a/internal/api/pull.go +++ b/internal/api/pull.go @@ -15,25 +15,25 @@ func (manager *ApiManagerCtx) pullStart(w http.ResponseWriter, r *http.Request) return } - err := manager.rooms.PullStart(request) + err := manager.pull.Start(request) if err != nil { http.Error(w, err.Error(), 500) return } - response := manager.rooms.PullStatus() + response := manager.pull.Status() w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(response) } func (manager *ApiManagerCtx) pullStatus(w http.ResponseWriter, r *http.Request) { - response := manager.rooms.PullStatus() + response := manager.pull.Status() w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(response) } func (manager *ApiManagerCtx) pullStop(w http.ResponseWriter, r *http.Request) { - err := manager.rooms.PullStop() + err := manager.pull.Stop() if err != nil { http.Error(w, err.Error(), 500) return diff --git a/internal/room/pull.go b/internal/pull/manager.go similarity index 51% rename from internal/room/pull.go rename to internal/pull/manager.go index a715e79..1f2074a 100644 --- a/internal/room/pull.go +++ b/internal/pull/manager.go @@ -1,4 +1,4 @@ -package room +package pull import ( "bufio" @@ -10,76 +10,73 @@ import ( "time" dockerTypes "github.com/docker/docker/api/types" + dockerClient "github.com/docker/docker/client" + "github.com/rs/zerolog" + "github.com/rs/zerolog/log" "github.com/m1k1o/neko-rooms/internal/types" "github.com/m1k1o/neko-rooms/internal/utils" ) -type PullConfig struct { +type PullManagerCtx struct { + logger zerolog.Logger + client *dockerClient.Client + images []string + mu sync.Mutex cancel func() status types.PullStatus layers map[string]int } -func (pull *PullConfig) TryInitialize(cancel func()) bool { - pull.mu.Lock() - defer pull.mu.Unlock() +func New(client *dockerClient.Client, nekoImages []string) *PullManagerCtx { + return &PullManagerCtx{ + logger: log.With().Str("module", "pull").Logger(), + client: client, + images: nekoImages, + } +} - if pull.status.Active { +func (manager *PullManagerCtx) tryInitialize(cancel func()) bool { + manager.mu.Lock() + defer manager.mu.Unlock() + + if manager.status.Active { cancel() return false } now := time.Now() - pull.cancel = cancel + manager.cancel = cancel - pull.status = types.PullStatus{ + manager.status = types.PullStatus{ Active: true, Started: &now, Layers: []types.PullLayer{}, Status: []string{}, } - pull.layers = map[string]int{} + manager.layers = map[string]int{} return true } -func (pull *PullConfig) SetDone() { - pull.mu.Lock() - defer pull.mu.Unlock() +func (manager *PullManagerCtx) setDone() { + manager.mu.Lock() + defer manager.mu.Unlock() now := time.Now() - pull.status.Active = false - pull.status.Finished = &now + manager.status.Active = false + manager.status.Finished = &now } -func (pull *PullConfig) Stop() bool { - pull.mu.Lock() - defer pull.mu.Unlock() - - if pull.status.Active { - pull.cancel() - } - - return pull.status.Active -} - -func (pull *PullConfig) Get() types.PullStatus { - pull.mu.Lock() - defer pull.mu.Unlock() - - return pull.status -} - -func (manager *RoomManagerCtx) PullStart(request types.PullStart) error { - if in, _ := utils.ArrayIn(request.NekoImage, manager.config.NekoImages); !in { +func (manager *PullManagerCtx) Start(request types.PullStart) error { + if in, _ := utils.ArrayIn(request.NekoImage, manager.images); !in { return fmt.Errorf("unknown neko image") } ctx, cancel := context.WithCancel(context.Background()) - if !manager.pull.TryInitialize(cancel) { + if !manager.tryInitialize(cancel) { return fmt.Errorf("pull is already in progess") } @@ -93,7 +90,7 @@ func (manager *RoomManagerCtx) PullStart(request types.PullStart) error { encodedJSON, err := json.Marshal(authConfig) if err != nil { - panic(err) + return err } opts = dockerTypes.ImagePullOptions{ @@ -104,7 +101,7 @@ func (manager *RoomManagerCtx) PullStart(request types.PullStart) error { reader, err := manager.client.ImagePull(ctx, request.NekoImage, opts) if err != nil { - manager.pull.SetDone() + manager.setDone() return err } @@ -115,52 +112,59 @@ func (manager *RoomManagerCtx) PullStart(request types.PullStart) error { layer := types.PullLayer{} if err := json.Unmarshal(data, &layer); err != nil { - manager.pull.status.Status = append( - manager.pull.status.Status, + manager.status.Status = append( + manager.status.Status, fmt.Sprintf("Error while parsing pull response: %s", err), ) - continue } if layer.ProgressDetail != nil { // map layer id to slice index - if index, ok := manager.pull.layers[layer.ID]; ok { - manager.pull.status.Layers[index] = layer + if index, ok := manager.layers[layer.ID]; ok { + manager.status.Layers[index] = layer } else { - manager.pull.layers[layer.ID] = len(manager.pull.layers) - manager.pull.status.Layers = append(manager.pull.status.Layers, layer) + manager.layers[layer.ID] = len(manager.layers) + manager.status.Layers = append(manager.status.Layers, layer) } } else { - manager.pull.status.Status = append( - manager.pull.status.Status, + manager.status.Status = append( + manager.status.Status, layer.Status, ) } } if err := scanner.Err(); err != nil { - manager.pull.status.Status = append( - manager.pull.status.Status, + manager.status.Status = append( + manager.status.Status, fmt.Sprintf("Error while reading pull response: %s", err), ) } reader.Close() - manager.pull.SetDone() + manager.setDone() }() return nil } -func (manager *RoomManagerCtx) PullStop() error { - if !manager.pull.Stop() { +func (manager *PullManagerCtx) Stop() error { + manager.mu.Lock() + defer manager.mu.Unlock() + + if !manager.status.Active { return fmt.Errorf("pull is not in progess") + } + manager.cancel() return nil } -func (manager *RoomManagerCtx) PullStatus() types.PullStatus { - return manager.pull.Get() +func (manager *PullManagerCtx) Status() types.PullStatus { + manager.mu.Lock() + defer manager.mu.Unlock() + + return manager.status } diff --git a/internal/room/manager.go b/internal/room/manager.go index d2f011a..efc87a7 100644 --- a/internal/room/manager.go +++ b/internal/room/manager.go @@ -34,20 +34,13 @@ const ( privateStorageGid = 1000 ) -func New(config *config.Room) *RoomManagerCtx { +func New(client *dockerClient.Client, config *config.Room) *RoomManagerCtx { logger := log.With().Str("module", "room").Logger() - cli, err := dockerClient.NewClientWithOpts(dockerClient.FromEnv) - if err != nil { - logger.Panic().Err(err).Msg("unable to connect to docker client") - } else { - logger.Info().Msg("successfully connected to docker client") - } - return &RoomManagerCtx{ logger: logger, config: config, - client: cli, + client: client, } } @@ -55,7 +48,6 @@ type RoomManagerCtx struct { logger zerolog.Logger config *config.Room client *dockerClient.Client - pull PullConfig } func (manager *RoomManagerCtx) Config() types.RoomsConfig { diff --git a/internal/types/pull.go b/internal/types/pull.go new file mode 100644 index 0000000..3037433 --- /dev/null +++ b/internal/types/pull.go @@ -0,0 +1,33 @@ +package types + +import "time" + +type PullStart struct { + NekoImage string `json:"neko_image"` + RegistryUser string `json:"registry_user"` + RegistryPass string `json:"registry_pass"` +} + +type PullLayer struct { + Status string `json:"status"` + ProgressDetail *struct { + Current int `json:"current"` + Total int `json:"total"` + } `json:"progressDetail"` + Progress string `json:"progress"` + ID string `json:"id"` +} + +type PullStatus struct { + Active bool `json:"active"` + Started *time.Time `json:"started"` + Layers []PullLayer `json:"layers"` + Status []string `json:"status"` + Finished *time.Time `json:"finished"` +} + +type PullManager interface { + Start(request PullStart) error + Stop() error + Status() PullStatus +} diff --git a/internal/types/room.go b/internal/types/room.go index 009616d..1290da9 100644 --- a/internal/types/room.go +++ b/internal/types/room.go @@ -246,30 +246,6 @@ type RoomMember struct { Muted bool `json:"muted"` } -type PullStart struct { - NekoImage string `json:"neko_image"` - RegistryUser string `json:"registry_user"` - RegistryPass string `json:"registry_pass"` -} - -type PullLayer struct { - Status string `json:"status"` - ProgressDetail *struct { - Current int `json:"current"` - Total int `json:"total"` - } `json:"progressDetail"` - Progress string `json:"progress"` - ID string `json:"id"` -} - -type PullStatus struct { - Active bool `json:"active"` - Started *time.Time `json:"started"` - Layers []PullLayer `json:"layers"` - Status []string `json:"status"` - Finished *time.Time `json:"finished"` -} - type RoomManager interface { Config() RoomsConfig List() ([]RoomEntry, error) @@ -284,8 +260,4 @@ type RoomManager interface { Start(id string) error Stop(id string) error Restart(id string) error - - PullStart(request PullStart) error - PullStop() error - PullStatus() PullStatus } diff --git a/neko.go b/neko.go index f7ca669..1acf664 100644 --- a/neko.go +++ b/neko.go @@ -6,6 +6,7 @@ import ( "os/signal" "runtime" + "github.com/docker/docker/client" "github.com/rs/zerolog" "github.com/rs/zerolog/log" "github.com/spf13/cobra" @@ -13,6 +14,7 @@ import ( "github.com/m1k1o/neko-rooms/internal/api" "github.com/m1k1o/neko-rooms/internal/config" "github.com/m1k1o/neko-rooms/internal/http" + "github.com/m1k1o/neko-rooms/internal/pull" "github.com/m1k1o/neko-rooms/internal/room" ) @@ -108,6 +110,7 @@ type MainCtx struct { logger zerolog.Logger roomManager *room.RoomManagerCtx + pullManager *pull.PullManagerCtx apiManager *api.ApiManagerCtx httpManager *http.HttpManagerCtx } @@ -117,12 +120,26 @@ func (main *MainCtx) Preflight() { } func (main *MainCtx) Start() { + client, err := client.NewClientWithOpts(client.FromEnv) + if err != nil { + main.logger.Panic().Err(err).Msg("unable to connect to docker client") + } else { + main.logger.Info().Msg("successfully connected to docker client") + } + main.roomManager = room.New( + client, main.Configs.Room, ) + main.pullManager = pull.New( + client, + main.Configs.Room.NekoImages, + ) + main.apiManager = api.New( main.roomManager, + main.pullManager, main.Configs.API, ) @@ -135,11 +152,8 @@ func (main *MainCtx) Start() { } func (main *MainCtx) Shutdown() { - if err := main.httpManager.Shutdown(); err != nil { - main.logger.Err(err).Msg("http manager shutdown with an error") - } else { - main.logger.Debug().Msg("http manager shutdown") - } + err := main.httpManager.Shutdown() + main.logger.Err(err).Msg("http manager shutdown") } func (main *MainCtx) ServeCommand(cmd *cobra.Command, args []string) { @@ -151,7 +165,7 @@ func (main *MainCtx) ServeCommand(cmd *cobra.Command, args []string) { signal.Notify(quit, os.Interrupt) sig := <-quit - main.logger.Warn().Msgf("received %s, attempting graceful shutdown: \n", sig) + main.logger.Warn().Msgf("received %s, attempting graceful shutdown.", sig) main.Shutdown() main.logger.Info().Msg("shutdown complete") }