package main import ( "embed" "fmt" "io" "log" "net/http" "os" "path/filepath" "time" "github.com/go-redis/redis/v8" "github.com/gorilla/mux" "github.com/hibiken/asynq" "github.com/rs/cors" ) // staticFileServer implements the http.Handler interface, so we can use it // to respond to HTTP requests. The path to the static directory and // path to the index file within that static directory are used to // serve the SPA in the given static directory. type staticFileServer struct { staticContents embed.FS staticDirPath string indexFileName string } // ServeHTTP inspects the URL path to locate a file within the static dir // on the SPA handler. // If path '/' is requested, it will serve the index file, otherwise it will // serve the file specified by the URL path. func (srv *staticFileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { // Get the absolute path to prevent directory traversal. path, err := filepath.Abs(r.URL.Path) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } if path == "/" { path = filepath.Join(srv.staticDirPath, srv.indexFileName) } else { path = filepath.Join(srv.staticDirPath, path) } f, err := srv.staticContents.Open(path) if err != nil { status := http.StatusInternalServerError if os.IsNotExist(err) { status = http.StatusNotFound } http.Error(w, err.Error(), status) return } defer f.Close() if _, err := io.Copy(w, f); err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } } const ( addr = "127.0.0.1:8080" redisAddr = "localhost:6379" // TODO: make this configurable ) //go:embed ui/build/* var staticContents embed.FS func main() { inspector := asynq.NewInspector(asynq.RedisClientOpt{ Addr: redisAddr, }) defer inspector.Close() rdb := redis.NewClient(&redis.Options{ Addr: redisAddr, }) defer rdb.Close() router := mux.NewRouter() router.Use(loggingMiddleware) api := router.PathPrefix("/api").Subrouter() // Queue endpoints. api.HandleFunc("/queues", newListQueuesHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}", newGetQueueHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}", newDeleteQueueHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}:pause", newPauseQueueHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}:resume", newResumeQueueHandlerFunc(inspector)).Methods("POST") // Queue Historical Stats endpoint. api.HandleFunc("/queue_stats", newListQueueStatsHandlerFunc(inspector)).Methods("GET") // Task endpoints. api.HandleFunc("/queues/{qname}/active_tasks", newListActiveTasksHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}/active_tasks/{task_id}:cancel", newCancelActiveTaskHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/active_tasks:cancel_all", newCancelAllActiveTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/active_tasks:batch_cancel", newBatchCancelActiveTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/pending_tasks", newListPendingTasksHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}/scheduled_tasks", newListScheduledTasksHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}/scheduled_tasks/{task_key}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/scheduled_tasks:delete_all", newDeleteAllScheduledTasksHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/scheduled_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks/{task_key}:run", newRunTaskHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks:run_all", newRunAllScheduledTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks:batch_run", newBatchRunTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks/{task_key}:kill", newKillTaskHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks:kill_all", newKillAllScheduledTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/scheduled_tasks:batch_kill", newBatchKillTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks", newListRetryTasksHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}/retry_tasks/{task_key}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/retry_tasks:delete_all", newDeleteAllRetryTasksHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/retry_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks/{task_key}:run", newRunTaskHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks:run_all", newRunAllRetryTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks:batch_run", newBatchRunTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks/{task_key}:kill", newKillTaskHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks:kill_all", newKillAllRetryTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/retry_tasks:batch_kill", newBatchKillTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/dead_tasks", newListDeadTasksHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/queues/{qname}/dead_tasks/{task_key}", newDeleteTaskHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/dead_tasks:delete_all", newDeleteAllDeadTasksHandlerFunc(inspector)).Methods("DELETE") api.HandleFunc("/queues/{qname}/dead_tasks:batch_delete", newBatchDeleteTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/dead_tasks/{task_key}:run", newRunTaskHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/dead_tasks:run_all", newRunAllDeadTasksHandlerFunc(inspector)).Methods("POST") api.HandleFunc("/queues/{qname}/dead_tasks:batch_run", newBatchRunTasksHandlerFunc(inspector)).Methods("POST") // Servers endpoints. api.HandleFunc("/servers", newListServersHandlerFunc(inspector)).Methods("GET") // Scheduler Entry endpoints. api.HandleFunc("/scheduler_entries", newListSchedulerEntriesHandlerFunc(inspector)).Methods("GET") api.HandleFunc("/scheduler_entries/{entry_id}/enqueue_events", newListSchedulerEnqueueEventsHandlerFunc(inspector)).Methods("GET") // Redis info endpoint. api.HandleFunc("/redis_info", newRedisInfoHandlerFunc(rdb)).Methods("GET") fs := &staticFileServer{ staticContents: staticContents, staticDirPath: "ui/build", indexFileName: "index.html", } router.PathPrefix("/").Handler(fs) c := cors.New(cors.Options{ AllowedMethods: []string{"GET", "POST", "DELETE"}, }) handler := c.Handler(router) srv := &http.Server{ Handler: handler, Addr: addr, WriteTimeout: 10 * time.Second, ReadTimeout: 10 * time.Second, } fmt.Printf("Asynq Monitoring WebUI server is running on %s\n", addr) log.Fatal(srv.ListenAndServe()) }