go-cart-actor/pkg/proxy/remotehost.go

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(500 * 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
}