package proxy import ( "context" "fmt" "io" "log" "net/http" "time" messages "git.tornberg.me/go-cart-actor/pkg/messages" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) // RemoteHost mirrors the lightweight controller used for remote node // interaction. type RemoteHost struct { Host string httpBase string conn *grpc.ClientConn transport *http.Transport client *http.Client controlClient messages.ControlPlaneClient MissedPings int } func NewRemoteHost(host string) (*RemoteHost, error) { target := fmt.Sprintf("%s:1337", host) conn, err := grpc.NewClient(target, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { log.Printf("AddRemote: dial %s failed: %v", target, err) return nil, err } controlClient := messages.NewControlPlaneClient(conn) for retries := range 3 { ctx, pingCancel := context.WithTimeout(context.Background(), time.Second) _, pingErr := controlClient.Ping(ctx, &messages.Empty{}) pingCancel() if pingErr == nil { break } if retries == 2 { log.Printf("AddRemote: ping %s failed after retries: %v", host, pingErr) conn.Close() return nil, pingErr } time.Sleep(200 * time.Millisecond) } transport := &http.Transport{ MaxIdleConns: 100, MaxIdleConnsPerHost: 100, IdleConnTimeout: 120 * time.Second, } client := &http.Client{Transport: transport, Timeout: 10 * time.Second} return &RemoteHost{ Host: host, httpBase: fmt.Sprintf("http://%s:8080/cart", host), conn: conn, transport: transport, client: client, controlClient: controlClient, MissedPings: 0, }, nil } func (h *RemoteHost) Name() string { return h.Host } func (h *RemoteHost) Close() error { if h.conn != nil { h.conn.Close() } return nil } func (h *RemoteHost) Ping() bool { ctx, cancel := context.WithTimeout(context.Background(), time.Second) _, err := h.controlClient.Ping(ctx, &messages.Empty{}) cancel() if err != nil { h.MissedPings++ log.Printf("Ping %s failed (%d) %v", h.Host, h.MissedPings, err) return false } h.MissedPings = 0 return true } func (h *RemoteHost) Negotiate(knownHosts []string) ([]string, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() resp, err := h.controlClient.Negotiate(ctx, &messages.NegotiateRequest{ KnownHosts: knownHosts, }) if err != nil { h.MissedPings++ log.Printf("Negotiate %s failed: %v", h.Host, err) return nil, err } h.MissedPings = 0 return resp.Hosts, nil } func (h *RemoteHost) GetActorIds() []uint64 { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() reply, err := h.controlClient.GetLocalActorIds(ctx, &messages.Empty{}) if err != nil { log.Printf("Init remote %s: GetCartIds error: %v", h.Host, err) h.MissedPings++ return []uint64{} } return reply.GetIds() } func (h *RemoteHost) AnnounceOwnership(uids []uint64) { _, err := h.controlClient.AnnounceOwnership(context.Background(), &messages.OwnershipAnnounce{ Host: h.Host, Ids: uids, }) if err != nil { log.Printf("ownership announce to %s failed: %v", h.Host, err) h.MissedPings++ return } h.MissedPings = 0 } func (h *RemoteHost) AnnounceExpiry(uids []uint64) { _, err := h.controlClient.AnnounceExpiry(context.Background(), &messages.ExpiryAnnounce{ Host: h.Host, Ids: uids, }) if err != nil { log.Printf("expiry announce to %s failed: %v", h.Host, err) h.MissedPings++ return } h.MissedPings = 0 } func (h *RemoteHost) Proxy(id uint64, w http.ResponseWriter, r *http.Request) (bool, error) { target := fmt.Sprintf("%s%s", h.httpBase, r.URL.RequestURI()) // var bodyCopy []byte // if r.Body != nil && r.Body != http.NoBody { // var err error // bodyCopy, err = io.ReadAll(r.Body) // if err != nil { // http.Error(w, "proxy read error", http.StatusBadGateway) // return false, err // } // } // if r.Body != nil { // r.Body.Close() // } // var reqBody io.Reader // if len(bodyCopy) > 0 { // reqBody = bytes.NewReader(bodyCopy) // } req, err := http.NewRequestWithContext(r.Context(), r.Method, target, r.Body) if err != nil { http.Error(w, "proxy build error", http.StatusBadGateway) return false, err } //r.Body = io.NopCloser(bytes.NewReader(bodyCopy)) req.Header.Set("X-Forwarded-Host", r.Host) for k, v := range r.Header { for _, vv := range v { req.Header.Add(k, vv) } } res, err := h.client.Do(req) if err != nil { http.Error(w, "proxy request error", http.StatusBadGateway) return false, err } defer res.Body.Close() for k, v := range res.Header { for _, vv := range v { w.Header().Add(k, vv) } } w.Header().Set("X-Cart-Owner-Routed", "true") if res.StatusCode >= 200 && res.StatusCode <= 299 { w.WriteHeader(res.StatusCode) _, copyErr := io.Copy(w, res.Body) if copyErr != nil { return true, copyErr } return true, nil } return false, fmt.Errorf("proxy response status %d", res.StatusCode) } func (r *RemoteHost) IsHealthy() bool { return r.MissedPings < 3 }