package discovery import ( "encoding/json" "errors" "fmt" "net/http" "slices" "strings" "time" apimeta "k8s.io/apimachinery/pkg/api/meta" "sigs.k8s.io/controller-runtime/pkg/client" crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/lease" ) // proxyView is the wire shape of one proxy in list and lease responses. type proxyView struct { ID string `json:"id"` // namespace/name IP string `json:"ip"` Port int32 `json:"port"` Attributes map[string]string `json:"attributes,omitempty"` Phase string `json:"phase"` Healthy bool `json:"healthy"` LatencyMillis int64 `json:"latencyMillis"` ActiveLeases int `json:"activeLeases"` MaxLeases int32 `json:"maxLeases"` } func viewOf(p *crawlv1alpha1.Proxy, activeLeases int) proxyView { return proxyView{ ID: client.ObjectKeyFromObject(p).String(), IP: p.EffectiveHost(), Port: p.EffectivePort(), Attributes: p.Spec.Attributes, Phase: string(p.Status.Phase), Healthy: isHealthy(p), LatencyMillis: p.Status.LatencyMillis, ActiveLeases: activeLeases, MaxLeases: p.MaxLeasesOrDefault(), } } func isHealthy(p *crawlv1alpha1.Proxy) bool { return apimeta.IsStatusConditionTrue(p.Status.Conditions, crawlv1alpha1.ConditionHealthy) } // matchesAttributes is spec.attributes equality: every selector pair must // be present verbatim. func matchesAttributes(p *crawlv1alpha1.Proxy, selector map[string]string) bool { for k, v := range selector { if p.Spec.Attributes[k] != v { return false } } return true } // GET /v1/proxies?attr.=&healthy=true|false func (s *Server) handleListProxies(w http.ResponseWriter, r *http.Request) { selector := map[string]string{} var healthyFilter *bool for key, values := range r.URL.Query() { switch { case key == "healthy": switch values[0] { case "true": healthyFilter = ptr(true) case "false": healthyFilter = ptr(false) default: writeError(w, http.StatusBadRequest, "invalid_query", fmt.Sprintf("healthy must be true or false, got %q", values[0])) return } case strings.HasPrefix(key, "attr."): selector[strings.TrimPrefix(key, "attr.")] = values[0] } } var list crawlv1alpha1.ProxyList if err := s.Reader.List(r.Context(), &list); err != nil { s.log.Error(err, "listing proxies") writeError(w, http.StatusInternalServerError, "internal", "listing proxies failed") return } counts := s.Store.Counts() views := []proxyView{} for i := range list.Items { p := &list.Items[i] if !p.DeletionTimestamp.IsZero() || !matchesAttributes(p, selector) { continue } v := viewOf(p, counts[client.ObjectKeyFromObject(p).String()]) if healthyFilter != nil && v.Healthy != *healthyFilter { continue } views = append(views, v) } slices.SortFunc(views, func(a, b proxyView) int { return strings.Compare(a.ID, b.ID) }) writeJSON(w, http.StatusOK, map[string]any{"proxies": views, "count": len(views)}) } type leaseRequest struct { Selector map[string]string `json:"selector"` TTLSeconds int64 `json:"ttlSeconds"` Target string `json:"target"` } type leaseResponse struct { LeaseID string `json:"leaseID"` Proxy proxyView `json:"proxy"` ExpiresAt time.Time `json:"expiresAt"` TTLSeconds int64 `json:"ttlSeconds"` } // POST /v1/leases func (s *Server) handleAcquireLease(w http.ResponseWriter, r *http.Request) { var req leaseRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "invalid_body", err.Error()) return } ttl := time.Duration(req.TTLSeconds) * time.Second if req.TTLSeconds == 0 { ttl = defaultTTL } if ttl < 0 || ttl > s.MaxLeaseTTL { writeError(w, http.StatusBadRequest, "invalid_ttl", fmt.Sprintf("ttlSeconds must be between 1 and %d", int64(s.MaxLeaseTTL.Seconds()))) return } var list crawlv1alpha1.ProxyList if err := s.Reader.List(r.Context(), &list); err != nil { s.log.Error(err, "listing proxies for lease") writeError(w, http.StatusInternalServerError, "internal", "listing proxies failed") return } // The store gets only healthy, live candidates; unhealthy matches are // counted here because the store never sees them. var candidates []lease.Candidate byKey := map[string]*crawlv1alpha1.Proxy{} unhealthy := 0 for i := range list.Items { p := &list.Items[i] if !p.DeletionTimestamp.IsZero() || !matchesAttributes(p, req.Selector) { continue } if !isHealthy(p) { unhealthy++ continue } key := client.ObjectKeyFromObject(p).String() byKey[key] = p candidates = append(candidates, lease.Candidate{ Proxy: key, MaxLeases: p.MaxLeasesOrDefault(), Latency: time.Duration(p.Status.LatencyMillis) * time.Millisecond, }) } granted, stats, err := s.Store.Acquire(r.Context(), lease.AcquireRequest{ Candidates: candidates, Target: req.Target, TTL: ttl, }) if errors.Is(err, lease.ErrNoMatch) { if s.Metrics != nil { s.Metrics.LeaseRequest("no_match") } writeJSON(w, http.StatusConflict, map[string]any{ "error": "no_match", "message": "no healthy proxy with free capacity matched the selector", "considered": stats.Considered + unhealthy, "atCapacity": stats.AtCapacity, "inCooldown": stats.InCooldown, "unhealthy": unhealthy, }) return } if err != nil { writeError(w, http.StatusBadRequest, "invalid_request", err.Error()) return } if s.Metrics != nil { s.Metrics.LeaseRequest("granted") } writeJSON(w, http.StatusCreated, leaseResponse{ LeaseID: granted.ID, Proxy: viewOf(byKey[granted.Proxy], s.Store.Counts()[granted.Proxy]), ExpiresAt: granted.ExpiresAt, TTLSeconds: int64(ttl.Seconds()), }) } // DELETE /v1/leases/{id} — early release, always 204: releasing an unknown // or already expired lease is not an error. func (s *Server) handleReleaseLease(w http.ResponseWriter, r *http.Request) { s.Store.Release(r.Context(), r.PathValue("id")) w.WriteHeader(http.StatusNoContent) } type reportRequest struct { Result string `json:"result"` Target string `json:"target"` } // POST /v1/leases/{id}/report func (s *Server) handleReportLease(w http.ResponseWriter, r *http.Request) { var req reportRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { writeError(w, http.StatusBadRequest, "invalid_body", err.Error()) return } result, ok := lease.ParseResult(req.Result) if !ok { writeError(w, http.StatusBadRequest, "invalid_result", fmt.Sprintf("result must be one of ok, rate_limited, banned; got %q", req.Result)) return } if err := s.Store.Report(r.Context(), r.PathValue("id"), result, req.Target); err != nil { if errors.Is(err, lease.ErrUnknownLease) { writeError(w, http.StatusNotFound, "unknown_lease", "no such lease") return } s.log.Error(err, "reporting lease", "leaseID", r.PathValue("id")) writeError(w, http.StatusInternalServerError, "internal", "report failed") return } w.WriteHeader(http.StatusNoContent) } func ptr[T any](v T) *T { return &v }