diff --git a/internal/proxy/manager.go b/internal/proxy/manager.go new file mode 100644 index 0000000..6bb3d30 --- /dev/null +++ b/internal/proxy/manager.go @@ -0,0 +1,256 @@ +package proxy + +import ( + "context" + "fmt" + "net/http" + "net/http/httputil" + "net/url" + "path" + "strings" + "sync" + + dockerTypes "github.com/docker/docker/api/types" + "github.com/docker/docker/api/types/filters" + dockerClient "github.com/docker/docker/client" + "github.com/rs/zerolog" + "github.com/rs/zerolog/log" + + "github.com/m1k1o/neko-rooms/internal/config" +) + +type ProxyManagerCtx struct { + logger zerolog.Logger + mu sync.RWMutex + ctx context.Context + cancel func() + prefix string + client *dockerClient.Client + config *config.Room + handlers map[string]http.Handler +} + +func New(client *dockerClient.Client, config *config.Room) *ProxyManagerCtx { + logger := log.With().Str("module", "proxy").Logger() + + prefix := config.PathPrefix + if !strings.HasSuffix(prefix, "/") { + prefix += "/" + } + + return &ProxyManagerCtx{ + logger: logger, + prefix: prefix, + client: client, + config: config, + handlers: map[string]http.Handler{}, + } +} + +func (p *ProxyManagerCtx) Start() { + p.ctx, p.cancel = context.WithCancel(context.Background()) + + go func() { + err := p.Refresh() + if err != nil { + p.logger.Err(err).Msg("unable to refresh containers") + } + + msgs, errs := p.client.Events(p.ctx, dockerTypes.EventsOptions{ + Filters: filters.NewArgs( + filters.Arg("type", "container"), + filters.Arg("label", fmt.Sprintf("m1k1o.neko_rooms.instance=%s", p.config.InstanceName)), + filters.Arg("event", "create"), + filters.Arg("event", "start"), + filters.Arg("event", "stop"), + filters.Arg("event", "destroy"), + ), + }) + + for { + select { + case err := <-errs: + p.logger.Info().Interface("err", err).Msg("eee") + case msg := <-msgs: + host := msg.ID[:12] + name, port, ok := p.parseLabels(msg.Actor.Attributes) + if !ok { + break + } + + switch msg.Action { + case "create": + p.logger.Info(). + Str("id", msg.ID). + Str("name", name). + Str("host", host). + Msg("container created") + + case "start": + p.Add(name, host+":"+port) + + p.logger.Info(). + Str("id", msg.ID). + Str("name", name). + Str("host", host). + Msg("container started") + + case "stop": + p.Remove(name) + + p.logger.Info(). + Str("id", msg.ID). + Str("name", name). + Str("host", host). + Msg("container stopped") + + case "destroy": + p.logger.Info(). + Str("id", msg.ID). + Str("name", name). + Str("host", host). + Msg("container destroyed") + } + } + } + }() +} + +func (p *ProxyManagerCtx) Shutdown() error { + p.Clear() + + p.cancel() + return p.ctx.Err() +} + +func (p *ProxyManagerCtx) refresh() error { + containers, err := p.client.ContainerList(p.ctx, dockerTypes.ContainerListOptions{ + All: true, + Filters: filters.NewArgs( + filters.Arg("label", fmt.Sprintf("m1k1o.neko_rooms.instance=%s", p.config.InstanceName)), + ), + }) + + if err != nil { + return err + } + + p.clear() + + for _, cont := range containers { + name, port, ok := p.parseLabels(cont.Labels) + if ok { + host := cont.ID[:12] + p.add(name, host+":"+port) + } + } + + return nil +} + +func (p *ProxyManagerCtx) parseLabels(labels map[string]string) (name string, port string, ok bool) { + name, ok = labels["m1k1o.neko_rooms.name"] + if !ok { + return + } + + // TODO: Do not use trafik for this. + port, ok = labels["traefik.http.services.neko-rooms-"+name+"-frontend.loadbalancer.server.port"] + return +} + +func (p *ProxyManagerCtx) add(name, host string) { + p.handlers[name] = httputil.NewSingleHostReverseProxy(&url.URL{ + Scheme: "http", + Host: host, + }) + + p.logger.Debug().Str("name", name).Str("host", host).Msg("add handler") +} + +func (p *ProxyManagerCtx) remove(name string) { + delete(p.handlers, name) + p.logger.Debug().Str("name", name).Msg("remove handler") +} + +func (p *ProxyManagerCtx) clear() { + p.handlers = map[string]http.Handler{} + p.logger.Debug().Msg("clear handlers") +} + +func (p *ProxyManagerCtx) Refresh() error { + p.mu.Lock() + defer p.mu.Unlock() + + return p.refresh() +} + +func (p *ProxyManagerCtx) Add(name, host string) { + p.mu.Lock() + defer p.mu.Unlock() + + p.add(name, host) +} + +func (p *ProxyManagerCtx) Remove(name string) { + p.mu.Lock() + defer p.mu.Unlock() + + p.remove(name) +} + +func (p *ProxyManagerCtx) Clear() { + p.mu.Lock() + defer p.mu.Unlock() + + p.clear() +} + +func (p *ProxyManagerCtx) ServeHTTP(w http.ResponseWriter, r *http.Request) { + // only if has prefix + if !strings.HasPrefix(r.URL.Path, p.prefix) { + http.NotFound(w, r) + return + } + + // remove prefix and leading / + roomPath := strings.TrimPrefix(r.URL.Path, p.prefix) + roomPath = strings.TrimLeft(roomPath, "/") + if roomPath == "" { + http.NotFound(w, r) + return + } + + // get room name + roomName, doRedir := roomPath, false + if i := strings.Index(roomPath, "/"); i != -1 { + roomName = roomPath[:i] + } else { + doRedir = true + } + + // get proxy by room name + p.mu.RLock() + proxy, ok := p.handlers[roomName] + p.mu.RUnlock() + + // if room not found + if !ok { + http.NotFound(w, r) + return + } + + // redirect to room ending with / + if doRedir { + r.URL.Path += "/" + http.Redirect(w, r, r.URL.String(), http.StatusTemporaryRedirect) + return + } + + // strip prefix from proxy + pathPrefix := path.Join(p.prefix, roomName) + proxy = http.StripPrefix(pathPrefix, proxy) + + // handle by proxy + proxy.ServeHTTP(w, r) +} diff --git a/internal/server/manager.go b/internal/server/manager.go index 5bd7e19..61ff432 100644 --- a/internal/server/manager.go +++ b/internal/server/manager.go @@ -24,7 +24,7 @@ type ServerManagerCtx struct { config *config.Server } -func New(ApiManager types.ApiManager, pathPrefix string, config *config.Server) *ServerManagerCtx { +func New(ApiManager types.ApiManager, pathPrefix string, config *config.Server, handler http.Handler) *ServerManagerCtx { logger := log.With().Str("module", "server").Logger() router := chi.NewRouter() @@ -73,6 +73,8 @@ func New(ApiManager types.ApiManager, pathPrefix string, config *config.Server) logger.Info().Msgf("with pprof endpoint at %s", pprofPath) } + router.Handle("/*", handler) + // we could use custom 404 router.NotFound(http.NotFound) diff --git a/internal/types/proxy.go b/internal/types/proxy.go new file mode 100644 index 0000000..f5b2cb0 --- /dev/null +++ b/internal/types/proxy.go @@ -0,0 +1,8 @@ +package types + +import "net/http" + +type ProxyManager interface { + ServeHTTP(w http.ResponseWriter, r *http.Request) + Shutdown() error +} diff --git a/neko.go b/neko.go index 07ca2c3..9543751 100644 --- a/neko.go +++ b/neko.go @@ -13,6 +13,7 @@ import ( "github.com/m1k1o/neko-rooms/internal/api" "github.com/m1k1o/neko-rooms/internal/config" + "github.com/m1k1o/neko-rooms/internal/proxy" "github.com/m1k1o/neko-rooms/internal/pull" "github.com/m1k1o/neko-rooms/internal/room" "github.com/m1k1o/neko-rooms/internal/server" @@ -112,6 +113,7 @@ type MainCtx struct { roomManager *room.RoomManagerCtx pullManager *pull.PullManagerCtx apiManager *api.ApiManagerCtx + proxyManager *proxy.ProxyManagerCtx serverManager *server.ServerManagerCtx } @@ -143,17 +145,29 @@ func (main *MainCtx) Start() { main.Configs.API, ) + main.proxyManager = proxy.New( + client, + main.Configs.Room, + ) + main.proxyManager.Start() + main.serverManager = server.New( main.apiManager, main.Configs.Room.PathPrefix, main.Configs.Server, + main.proxyManager, ) main.serverManager.Start() } func (main *MainCtx) Shutdown() { - err := main.serverManager.Shutdown() + var err error + + err = main.serverManager.Shutdown() main.logger.Err(err).Msg("server manager shutdown") + + err = main.proxyManager.Shutdown() + main.logger.Err(err).Msg("proxy manager shutdown") } func (main *MainCtx) ServeCommand(cmd *cobra.Command, args []string) {