boltrouter

Documentation

Index

• Constants
• Variables
• func CopyReqBody(src *http.Request, dest *http.Request)
• func CopyRespBody(resp *http.Response) io.ReadCloser
• func StatusCodeIs2xx(statusCode int) bool
• type AtomicVar
  • func (v *AtomicVar[T]) Get() T
  • func (v *AtomicVar[T]) Set(x T)
  • func (v *AtomicVar[T]) String() string
• type BoltInfo
• type BoltRequest
• type BoltRequestAnalytics
• type BoltRouter
  • func NewBoltRouter(ctx context.Context, logger *zap.Logger, cfg Config) (*BoltRouter, error)
  • func (br *BoltRouter) DoRequest(logger *zap.Logger, boltReq *BoltRequest) (*http.Response, bool, *BoltRequestAnalytics, error)
  • func (br *BoltRouter) GetCleanerStatus() (bool, error)
  • func (br *BoltRouter) IsOffline(endpoint string) bool
  • func (br *BoltRouter) MaybeMarkOffline(url *url.URL, err error)
  • func (br *BoltRouter) NewBoltRequest(ctx context.Context, logger *zap.Logger, req *http.Request) (*BoltRequest, error)
  • func (br *BoltRouter) RefreshAWSCredentialsPeriodically(ctx context.Context, logger *zap.Logger)
  • func (br *BoltRouter) RefreshBoltInfo(ctx context.Context) error
  • func (br *BoltRouter) RefreshBoltInfoPeriodically(ctx context.Context)
  • func (br *BoltRouter) SelectBoltEndpoint(reqMethod string) (*url.URL, error)
  • func (br *BoltRouter) SelectInitialRequestTarget(boltReq *BoltRequest) (target InitialRequestTargetType, reason string, err error)
• type BoltVars
  • func GetBoltVars(ctx context.Context, logger *zap.Logger, cloudPlatform CloudPlatformType) (*BoltVars, error)
  • func (bv *BoltVars) MarshalLogObject(enc zapcore.ObjectEncoder) error
• type CloudPlatformType
• type Config
• type CrunchTrafficSplitType
• type ErrUnknownCloudPlatform
• type InitialRequestTargetType
• type SourceBucket

Constants

const (
	CrunchTrafficSplitByObjectKeyHash CrunchTrafficSplitType = "objectkeyhash"
	CrunchTrafficSplitByRandomRequest CrunchTrafficSplitType = "random"
	UndefinedCloudPlatform            CloudPlatformType      = 0
	AwsCloudPlatform                  CloudPlatformType      = 1
	GcpCloudPlatform                  CloudPlatformType      = 2
)

const (
	BoltInfoRefreshInterval = 10 * time.Second
)

Variables

var (
	ErrPanicDuringBoltRequest = errors.New("panic occurred during Bolt request")
	ErrPanicDuringAwsRequest  = errors.New("panic occurred during AWS request")
)

var (
	CloudPlatformStrToTypeMap map[string]CloudPlatformType = map[string]CloudPlatformType{
		"aws": AwsCloudPlatform,
		"gcp": GcpCloudPlatform,
	}
	CloudPlatformTypeToStrMap map[CloudPlatformType]string = map[CloudPlatformType]string{
		UndefinedCloudPlatform: "undefined",
		AwsCloudPlatform:       "aws",
		GcpCloudPlatform:       "gcp",
	}
)

var DefaultConfig = Config{
	Local:                     false,
	CloudPlatform:             UndefinedCloudPlatform,
	Passthrough:               false,
	Failover:                  false,
	NoFallback404:             false,
	BoltEndpointOverride:      "",
	CrunchTrafficSplit:        CrunchTrafficSplitByObjectKeyHash,
	GcpReplicasEnabled:        false,
	AwsIgnoreAuthHeaderRegion: false,
}

var InitialRequestTargetMap = map[InitialRequestTargetType]string{
	InitialRequestTargetUndefined: "undefined",
	InitialRequestTargetBolt:      "bolt",
	InitialRequestTargetFallback:  "fallback",
}

Functions

func CopyReqBody

func CopyReqBody(src *http.Request, dest *http.Request)

CopyReqBody copies the request body into a destination request. this allows reading a request body multiple times without "closing" it

func CopyRespBody

func CopyRespBody(resp *http.Response) io.ReadCloser

CopyRespBody copies the response body and returns a new response with the copied body. this allows reading a response body multiple times without "closing" it.

func StatusCodeIs2xx

func StatusCodeIs2xx(statusCode int) bool

Types

type AtomicVar

type AtomicVar[T any] struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

AtomicVar is a generic thread safe wrapper for variables of any type

func (*AtomicVar[T]) Get

func (v *AtomicVar[T]) Get() T

Get uses sync.RLock to access the variable

func (*AtomicVar[T]) Set

func (v *AtomicVar[T]) Set(x T)

Set uses sync.Lock to write to the variable

func (*AtomicVar[T]) String

func (v *AtomicVar[T]) String() string

type BoltInfo

type BoltInfo map[string]interface{}

{
	"main_write_endpoints": [],
	"failover_write_endpoints": [],
	"main_read_endpoints": [],
	"failover_read_endpoints": [],
	"cluster_healthy": bool,
	"client_behavior_params": {
		"cleaner_on": bool
		"crunch_traffic_percent": int
	}
}

type BoltRequest

type BoltRequest struct {
	Bolt *http.Request
	Aws  *http.Request
	Gcp  *http.Request
	// contains filtered or unexported fields
}

type BoltRequestAnalytics

type BoltRequestAnalytics struct {
	ObjectKey                     string
	RequestBodySize               uint32
	Method                        string
	InitialRequestTarget          string
	InitialRequestTargetReason    string
	BoltRequestUrl                string
	BoltRequestDuration           time.Duration
	BoltRequestResponseStatusCode int
	AwsRequestDuration            time.Duration
	AwsRequestResponseStatusCode  int
}

type BoltRouter

type BoltRouter struct {
	// contains filtered or unexported fields
}

BoltRouter is used to find bolt endpoints and route a AWS call to the right endpoint.

func NewBoltRouter

func NewBoltRouter(ctx context.Context, logger *zap.Logger, cfg Config) (*BoltRouter, error)

NewBoltRouter creates a new BoltRouter.

func (*BoltRouter) DoRequest

func (br *BoltRouter) DoRequest(logger *zap.Logger, boltReq *BoltRequest) (*http.Response, bool, *BoltRequestAnalytics, error)

DoRequest sends an HTTP Bolt request and returns an HTTP response, following policy (such as redirects, cookies, auth) as configured on the client. DoRequest will failover to AWS if the Bolt request fails and the config.Failover is set to true. DoRequest will failover to AWS if the Bolt request panics for any reason DoboltRequest will return a bool indicating if the request was a failover. DoRequest will return a BoltRequestAnalytics struct with analytics about the request.

func (*BoltRouter) GetCleanerStatus

func (br *BoltRouter) GetCleanerStatus() (bool, error)

func (*BoltRouter) IsOffline

func (br *BoltRouter) IsOffline(endpoint string) bool

IsOffline is called to check if a bolt endpoint is offline

func (*BoltRouter) MaybeMarkOffline

func (br *BoltRouter) MaybeMarkOffline(url *url.URL, err error)

MaybeMarkOffline is called to mark bolt endpoints offline on error.

func (*BoltRouter) NewBoltRequest

func (br *BoltRouter) NewBoltRequest(ctx context.Context, logger *zap.Logger, req *http.Request) (*BoltRequest, error)

NewBoltRequest transforms the passed in intercepted aws or gcp http.Request and returns a new http.Request Ready to be sent to Bolt. This new http.Request is routed to the correct Bolt endpoint and signed correctly.

func (*BoltRouter) RefreshAWSCredentialsPeriodically

func (br *BoltRouter) RefreshAWSCredentialsPeriodically(ctx context.Context, logger *zap.Logger)

func (*BoltRouter) RefreshBoltInfo

func (br *BoltRouter) RefreshBoltInfo(ctx context.Context) error

RefreshBoltInfo refreshes the BoltVars.BoltInfo variable and restarts the refresh interval. Call this method to force refresh BoltVars.BoltInfo.

func (*BoltRouter) RefreshBoltInfoPeriodically

func (br *BoltRouter) RefreshBoltInfoPeriodically(ctx context.Context)

RefreshBoltInfoPeriodically starts a goroutine that calls RefreshBoltInfo every BoltInfoRefreshInterval seconds

func (*BoltRouter) SelectBoltEndpoint

func (br *BoltRouter) SelectBoltEndpoint(reqMethod string) (*url.URL, error)

SelectBoltEndpoint selects a bolt endpoint from BoltVars.BoltEndpoints from the passed in reqMethod. This method will err if not endpoints were selected.

func (*BoltRouter) SelectInitialRequestTarget

func (br *BoltRouter) SelectInitialRequestTarget(boltReq *BoltRequest) (target InitialRequestTargetType, reason string, err error)

select initial request destination based on cluster_health_metrics and client_behavior_params

type BoltVars

type BoltVars struct {
	ReadOrderEndpoints  AtomicVar[[]string]
	WriteOrderEndpoints AtomicVar[[]string]
	HttpReadMethodTypes AtomicVar[[]string]

	Region           AtomicVar[string]
	ZoneId           AtomicVar[string]
	BoltCustomDomain AtomicVar[string]
	AuthBucket       AtomicVar[string]
	UserAgentPrefix  AtomicVar[string]
	BoltHostname     AtomicVar[string]
	QuicksilverURL   AtomicVar[string]
	BoltInfo         AtomicVar[BoltInfo]
	// contains filtered or unexported fields
}

BoltVars is a singleton struct keeping track of Bolt variables across threads. This is used in BoltRouter to route requests appropriately. You should only access BoltVars with GetBoltVars().

func GetBoltVars

func GetBoltVars(ctx context.Context, logger *zap.Logger, cloudPlatform CloudPlatformType) (*BoltVars, error)

GetBoltVars acts as a singleton method wrapper around BoltVars. It guarantees that only one instance of BoltVars exists. This method is thread safe.

func (*BoltVars) MarshalLogObject

func (bv *BoltVars) MarshalLogObject(enc zapcore.ObjectEncoder) error

type CloudPlatformType

type CloudPlatformType uint8

type Config

type Config struct {
	// If set, boltrouter will run in local mode.
	// For example, it will not query quicksilver to get endpoints.
	Local bool `yaml:"Local"`

	// Set the cloud platform that Crunch is running in.
	CloudPlatform CloudPlatformType `yaml:"CloudPlatform"`

	// Set the BoltEndpointOverride while running from local mode.
	BoltEndpointOverride string `yaml:"BoltEndpointOverride"`

	// Enable pass through in Bolt.
	Passthrough bool `yaml:"Passthrough"`

	// Enable failover to a AWS request if the Bolt request fails
	Failover bool `yaml:"Failover"`

	// Enable NoFallback404 to disable fallback on 404 response code from AWS request to Bolt or vice-versa.
	// Fallback is useful on GetObject, where object maybe present in the other source.
	NoFallback404 bool `yaml:"NoFallback404"`

	// There are two ways to split the traffic between bolt and object store
	// 1. Random Crunch Traffic Split
	// 2. Hash Based Crunch Traffic Split
	// Random approach could cause data inconsistency if the requests are mix of GET and PUT.
	CrunchTrafficSplit CrunchTrafficSplitType `yaml:"CrunchTrafficSplit"`

	// Whether a GCP deployment is single endpoint or we have replicas to take advantage of.
	GcpReplicasEnabled bool `yaml:"GcpReplicasEnabled"`

	AwsIgnoreAuthHeaderRegion bool `yaml:"AwsIgnoreAuthHeaderRegion"`
}

type CrunchTrafficSplitType

type CrunchTrafficSplitType string

type ErrUnknownCloudPlatform

type ErrUnknownCloudPlatform error

type InitialRequestTargetType

type InitialRequestTargetType uint8

const (
	InitialRequestTargetUndefined InitialRequestTargetType = 0
	InitialRequestTargetBolt      InitialRequestTargetType = 1
	InitialRequestTargetFallback  InitialRequestTargetType = 2
)

type SourceBucket

type SourceBucket struct {
	Bucket string
	Region string
	Style  s3RequestStyle
}