242 lines
7.2 KiB
Go
242 lines
7.2 KiB
Go
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.<key>=<value>&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 }
|