director.go

// Package http provides the http implementation of the Director interface.
package http

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"time"

	"github.com/featurebasedb/featurebase/v3/dax"
	"github.com/featurebasedb/featurebase/v3/dax/controller"
	"github.com/featurebasedb/featurebase/v3/errors"
	"github.com/featurebasedb/featurebase/v3/logger"
)

// Ensure type implements interface.
var _ controller.Director = (*Director)(nil)

// Director is an http implementation of the Director interface.
type Director struct {
	// directivePath is the path portion of the URI to which directives should
	// be POSTed.
	directivePath string

	// snapshotRequestPath is the path portion of the URI to which snapshot
	// requests should be POSTed.
	snapshotRequestPath string

	client *http.Client

	logger logger.Logger
}

func NewDirector(cfg DirectorConfig) *Director {
	var logr = logger.NopLogger
	if cfg.Logger != nil {
		logr = cfg.Logger
	}

	return &Director{
		directivePath:       cfg.DirectivePath,
		snapshotRequestPath: cfg.SnapshotRequestPath,
		logger:              logr,
		client: &http.Client{
			Transport: &http.Transport{
				Proxy: http.ProxyFromEnvironment,
				DialContext: (&net.Dialer{
					Timeout:   2 * time.Second,
					KeepAlive: 30 * time.Second,
				}).DialContext,
				ForceAttemptHTTP2:     true,
				MaxIdleConns:          100,
				IdleConnTimeout:       90 * time.Second,
				TLSHandshakeTimeout:   3 * time.Second,
				ExpectContinueTimeout: 1 * time.Second,
			},
		},
	}
}

type DirectorConfig struct {
	DirectivePath       string
	SnapshotRequestPath string
	Logger              logger.Logger
}

func (d *Director) SendDirective(ctx context.Context, dir *dax.Directive) error {
	url := fmt.Sprintf("%s/%s", dir.Address.WithScheme("http"), d.directivePath)
	d.logger.Printf("SEND HTTP directive to: %s\n", url)
	d.logger.Debugf("Directive: %+v", dir)

	// Encode the request.
	postBody, err := json.Marshal(dir)
	if err != nil {
		return errors.Wrap(err, "marshalling directive to json")
	}
	requestBody := bytes.NewBuffer(postBody)

	// Post the request.
	request, _ := http.NewRequest(http.MethodPost, url, requestBody)
	request.Header.Add("Content-Type", "application/json")
	request.Header.Add("Accept", "application/json")

	resp, err := d.client.Do(request)
	if err != nil {
		return errors.Wrap(err, "doing send directive")
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
	}

	return nil
}

func (d *Director) SendSnapshotShardDataRequest(ctx context.Context, req *dax.SnapshotShardDataRequest) error {
	url := fmt.Sprintf("%s/%s/shard-data", req.Address.WithScheme("http"), d.snapshotRequestPath)

	// Encode the request.
	postBody, err := json.Marshal(req)
	if err != nil {
		return errors.Wrap(err, "marshalling snapshot shard data request to json")
	}
	requestBody := bytes.NewBuffer(postBody)

	// Post the request.
	request, _ := http.NewRequest(http.MethodPost, url, requestBody)
	request.Header.Add("Content-Type", "application/json")
	request.Header.Add("Accept", "application/json")

	resp, err := d.client.Do(request)
	if err != nil {
		return errors.Wrap(err, "doing snapshot shard data request")
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
	}

	return nil
}

func (d *Director) SendSnapshotTableKeysRequest(ctx context.Context, req *dax.SnapshotTableKeysRequest) error {
	url := fmt.Sprintf("%s/%s/table-keys", req.Address.WithScheme("http"), d.snapshotRequestPath)
	d.logger.Printf("SEND HTTP snapshot table keys request to: %s\n", url)

	// Encode the request.
	postBody, err := json.Marshal(req)
	if err != nil {
		return errors.Wrap(err, "marshalling snapshot table keys request to json")
	}
	requestBody := bytes.NewBuffer(postBody)

	// Post the request.
	request, _ := http.NewRequest(http.MethodPost, url, requestBody)
	request.Header.Add("Content-Type", "application/json")
	request.Header.Add("Accept", "application/json")

	resp, err := d.client.Do(request)
	if err != nil {
		return errors.Wrap(err, "doing snapshot table keys request")
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
	}

	return nil
}

func (d *Director) SendSnapshotFieldKeysRequest(ctx context.Context, req *dax.SnapshotFieldKeysRequest) error {
	url := fmt.Sprintf("%s/%s/field-keys", req.Address.WithScheme("http"), d.snapshotRequestPath)
	d.logger.Printf("SEND HTTP snapshot field keys request to: %s\n", url)

	// Encode the request.
	postBody, err := json.Marshal(req)
	if err != nil {
		return errors.Wrap(err, "marshalling snapshot field keys request to json")
	}
	requestBody := bytes.NewBuffer(postBody)

	// Post the request.
	request, _ := http.NewRequest(http.MethodPost, url, requestBody)
	request.Header.Add("Content-Type", "application/json")
	request.Header.Add("Accept", "application/json")

	resp, err := d.client.Do(request)
	if err != nil {
		return errors.Wrap(err, "doing snapshot field keys request")
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		b, _ := io.ReadAll(resp.Body)
		return errors.Errorf("status code: %d: %s", resp.StatusCode, b)
	}

	return nil
}