Enable to run taggers from within admin web interface

internal/api/rest.go

@@ -22,6 +22,7 @@ import (
 	"github.com/ClusterCockpit/cc-backend/internal/auth"
 	"github.com/ClusterCockpit/cc-backend/internal/config"
 	"github.com/ClusterCockpit/cc-backend/internal/repository"
+	"github.com/ClusterCockpit/cc-backend/internal/tagger"
 	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
 	"github.com/ClusterCockpit/cc-lib/v2/schema"
 	"github.com/ClusterCockpit/cc-lib/v2/util"
@@ -152,6 +153,8 @@ func (api *RestAPI) MountConfigAPIRoutes(r chi.Router) {
 		r.Delete("/config/users/", api.deleteUser)
 		r.Post("/config/user/{id}", api.updateUser)
 		r.Post("/config/notice/", api.editNotice)
+		r.Get("/config/taggers/", api.getTaggers)
+		r.Post("/config/taggers/run/", api.runTagger)
 	}
 }
 
@@ -268,6 +271,42 @@ func (api *RestAPI) editNotice(rw http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (api *RestAPI) getTaggers(rw http.ResponseWriter, r *http.Request) {
+	if user := repository.GetUserFromContext(r.Context()); !user.HasRole(schema.RoleAdmin) {
+		handleError(fmt.Errorf("only admins are allowed to list taggers"), http.StatusForbidden, rw)
+		return
+	}
+
+	rw.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(rw).Encode(tagger.ListTaggers()); err != nil {
+		cclog.Errorf("Failed to encode tagger list: %v", err)
+	}
+}
+
+func (api *RestAPI) runTagger(rw http.ResponseWriter, r *http.Request) {
+	if user := repository.GetUserFromContext(r.Context()); !user.HasRole(schema.RoleAdmin) {
+		handleError(fmt.Errorf("only admins are allowed to run taggers"), http.StatusForbidden, rw)
+		return
+	}
+
+	name := r.FormValue("name")
+	if name == "" {
+		handleError(fmt.Errorf("missing required parameter: name"), http.StatusBadRequest, rw)
+		return
+	}
+
+	if err := tagger.RunTaggerByName(name); err != nil {
+		handleError(err, http.StatusConflict, rw)
+		return
+	}
+
+	rw.Header().Set("Content-Type", "text/plain")
+	rw.WriteHeader(http.StatusOK)
+	if _, err := rw.Write([]byte(fmt.Sprintf("Tagger %s started", name))); err != nil {
+		cclog.Errorf("Failed to write response: %v", err)
+	}
+}
+
 // getJWT godoc
 // @summary     Generate JWT token
 // @tags        Frontend

internal/tagger/tagger.go

@@ -10,6 +10,7 @@
 package tagger
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/ClusterCockpit/cc-backend/internal/repository"
@@ -29,11 +30,31 @@ type Tagger interface {
 	Match(job *schema.Job)
 }
 
+// TaggerInfo holds metadata about a tagger for JSON serialization.
+type TaggerInfo struct {
+	Name    string `json:"name"`
+	Type    string `json:"type"`
+	Running bool   `json:"running"`
+}
+
 var (
-	initOnce  sync.Once
-	jobTagger *JobTagger
+	initOnce     sync.Once
+	jobTagger    *JobTagger
+	statusMu     sync.Mutex
+	taggerStatus = map[string]bool{}
 )
 
+// Known tagger definitions: name -> (type, factory)
+type taggerDef struct {
+	ttype   string
+	factory func() Tagger
+}
+
+var knownTaggers = map[string]taggerDef{
+	"AppTagger":      {ttype: "start", factory: func() Tagger { return &AppTagger{} }},
+	"JobClassTagger": {ttype: "stop", factory: func() Tagger { return &JobClassTagger{} }},
+}
+
 // JobTagger coordinates multiple taggers that run at different job lifecycle events.
 // It maintains separate lists of taggers that run when jobs start and when they stop.
 type JobTagger struct {
@@ -88,6 +109,73 @@ func (jt *JobTagger) JobStopCallback(job *schema.Job) {
 	}
 }
 
+// ListTaggers returns information about all known taggers with their current running status.
+func ListTaggers() []TaggerInfo {
+	statusMu.Lock()
+	defer statusMu.Unlock()
+
+	result := make([]TaggerInfo, 0, len(knownTaggers))
+	for name, def := range knownTaggers {
+		result = append(result, TaggerInfo{
+			Name:    name,
+			Type:    def.ttype,
+			Running: taggerStatus[name],
+		})
+	}
+	return result
+}
+
+// RunTaggerByName starts a tagger by name asynchronously on all jobs.
+// Returns an error if the name is unknown or the tagger is already running.
+func RunTaggerByName(name string) error {
+	def, ok := knownTaggers[name]
+	if !ok {
+		return fmt.Errorf("unknown tagger: %s", name)
+	}
+
+	statusMu.Lock()
+	if taggerStatus[name] {
+		statusMu.Unlock()
+		return fmt.Errorf("tagger %s is already running", name)
+	}
+	taggerStatus[name] = true
+	statusMu.Unlock()
+
+	go func() {
+		defer func() {
+			statusMu.Lock()
+			taggerStatus[name] = false
+			statusMu.Unlock()
+		}()
+
+		t := def.factory()
+		if err := t.Register(); err != nil {
+			cclog.Errorf("Failed to register tagger %s: %s", name, err)
+			return
+		}
+
+		r := repository.GetJobRepository()
+		jl, err := r.GetJobList(0, 0)
+		if err != nil {
+			cclog.Errorf("Error getting job list for tagger %s: %s", name, err)
+			return
+		}
+
+		cclog.Infof("Running tagger %s on %d jobs", name, len(jl))
+		for _, id := range jl {
+			job, err := r.FindByIDDirect(id)
+			if err != nil {
+				cclog.Errorf("Error getting job %d for tagger %s: %s", id, name, err)
+				continue
+			}
+			t.Match(job)
+		}
+		cclog.Infof("Tagger %s completed", name)
+	}()
+
+	return nil
+}
+
 // RunTaggers applies all configured taggers to all existing jobs in the repository.
 // This is useful for retroactively applying tags to jobs that were created before
 // the tagger system was initialized or when new tagging rules are added.