activator

README

About the Activator

The name activator is actually a misnomer, since after Knative 0.2, the activator no longer activates inactive Revisions.

The only responsibilities of the activator are:

• Receiving & buffering requests for inactive Revisions.
• Reporting metrics to the autoscaler.
• Retrying requests to a Revision after the autoscaler scales such Revision based on the reported metrics.

Documentation

Index

• Constants
• Variables
• func DeleteBreaker(throttler *Throttler) func(obj interface{})
• func EndpointsAddressCount(subsets []corev1.EndpointSubset) int
• func ServicePort(protocol v1alpha1.RevisionProtocolType) int32
• func UpdateEndpoints(throttler *Throttler) func(newObj interface{})
• type ActivationResult
• type Activator
  • func NewDedupingActivator(a Activator) Activator
  • func NewRevisionActivator(kubeClient kubernetes.Interface, servingClient clientset.Interface, ...) Activator
• type Endpoint
• type Reporter
  • func NewStatsReporter() (*Reporter, error)
  • func (r *Reporter) ReportRequestCount(ns, service, config, rev string, responseCode, numTries int, v int64) error
  • func (r *Reporter) ReportResponseTime(ns, service, config, rev string, responseCode int, d time.Duration) error
• type RevisionID
• type StatsReporter
• type Throttler
  • func NewThrottler(params ThrottlerParams) *Throttler
  • func (t *Throttler) Remove(rev RevisionID)
  • func (t *Throttler) Try(rev RevisionID, function func()) error
  • func (t *Throttler) UpdateCapacity(rev RevisionID, size int32) error
• type ThrottlerParams

Constants

const (
	// K8sServiceName is the name of the activator service
	K8sServiceName = "activator-service"
	// RequestCountHTTPHeader is the header key for number of tries
	RequestCountHTTPHeader string = "knative-activator-num-retries"
	// RevisionHeaderName is the header key for revision name
	RevisionHeaderName string = "knative-serving-revision"
	// RevisionHeaderNamespace is the header key for revision's namespace
	RevisionHeaderNamespace string = "knative-serving-namespace"

	// ServicePortHTTP1 is the port number for activating HTTP1 revisions
	ServicePortHTTP1 int32 = 80
	// ServicePortHTTP1 is the port number for activating H2C revisions
	ServicePortH2C int32 = 81
)

Variables

var ErrActivatorOverload = errors.New("activator overload")

ErrActivatorOverload indicates that throttler has no free slots to buffer the request.

Functions

func DeleteBreaker

func DeleteBreaker(throttler *Throttler) func(obj interface{})

DeleteBreaker is a handler function to be used by the Endpoints informer. It removes the Breaker from the Throttler bookkeeping.

func EndpointsAddressCount

func EndpointsAddressCount(subsets []corev1.EndpointSubset) int

EndpointsAddressCount returns the total number of addresses registered for the endpoint.

func ServicePort

func ServicePort(protocol v1alpha1.RevisionProtocolType) int32

ServicePort returns the activator service port for the given Revision protocol. Default is `ServicePortHTTP1`.

func UpdateEndpoints

func UpdateEndpoints(throttler *Throttler) func(newObj interface{})

UpdateEndpoints is a handler function to be used by the Endpoints informer. It updates the endpoints in the Throttler if the number of hosts changed and the revision already exists (we don't want to create/update throttlers for the endpoints that do not belong to any revision).

This function must not be called in parallel to not induce a wrong order of events.

Types

type ActivationResult

type ActivationResult struct {
	Status            int
	Endpoint          Endpoint
	ServiceName       string
	ConfigurationName string
	Error             error
}

ActivationResult is used to return the result of an ActivateEndpoint call

type Activator

type Activator interface {
	ActiveEndpoint(namespace, name string) ActivationResult
	Shutdown()
}

Activator provides an active endpoint for a revision or an error and status code indicating why it could not.

func NewDedupingActivator

func NewDedupingActivator(a Activator) Activator

NewDedupingActivator creates an Activator that deduplicates activations requests for the same revision id and namespace.

func NewRevisionActivator

func NewRevisionActivator(kubeClient kubernetes.Interface, servingClient clientset.Interface, logger *zap.SugaredLogger) Activator

NewRevisionActivator creates an Activator that changes revision serving status to active if necessary, then returns the endpoint once the revision is ready to serve traffic.

type Endpoint

type Endpoint struct {
	FQDN string
	Port int32
}

Endpoint is a fully-qualified domain name / port pair for an active revision.

type Reporter

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

Reporter holds cached metric objects to report autoscaler metrics

func NewStatsReporter

func NewStatsReporter() (*Reporter, error)

NewStatsReporter creates a reporter that collects and reports activator metrics

func (*Reporter) ReportRequestCount

func (r *Reporter) ReportRequestCount(ns, service, config, rev string, responseCode, numTries int, v int64) error

ReportRequestCount captures request count metric with value v.

func (*Reporter) ReportResponseTime

func (r *Reporter) ReportResponseTime(ns, service, config, rev string, responseCode int, d time.Duration) error

ReportResponseTime captures response time requests

type RevisionID

type RevisionID struct {
	Namespace string
	Name      string
}

RevisionID is the combination of namespace and service name

type StatsReporter

type StatsReporter interface {
	ReportRequestCount(ns, service, config, rev string, responseCode, numTries int, v int64) error
	ReportResponseTime(ns, service, config, rev string, responseCode int, d time.Duration) error
}

StatsReporter defines the interface for sending activator metrics

type Throttler

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

Throttler keeps the mapping of Revisions to Breakers and allows updating max concurrency dynamically of respective Breakers. Max concurrency is essentially the number of semaphore tokens the Breaker has in rotation. The manipulation of the parameter is done via `UpdateCapacity()` method. It enables the use case to start with max concurrency set to 0 (no requests are sent because no endpoints are available) and gradually increase its value depending on the external condition (e.g. new endpoints become available)

func NewThrottler

func NewThrottler(params ThrottlerParams) *Throttler

NewThrottler creates a new Throttler.

func (*Throttler) Remove

func (t *Throttler) Remove(rev RevisionID)

Remove deletes the breaker from the bookkeeping.

func (*Throttler) Try

func (t *Throttler) Try(rev RevisionID, function func()) error

Try potentially registers a new breaker in our bookkeeping and executes the `function` on the Breaker. It returns an error if either breaker doesn't have enough capacity, or breaker's registration didn't succeed, e.g. getting endpoints or update capacity failed.

func (*Throttler) UpdateCapacity

func (t *Throttler) UpdateCapacity(rev RevisionID, size int32) error

UpdateCapacity updates the max concurrency of the Breaker corresponding to a revision.

type ThrottlerParams

type ThrottlerParams struct {
	BreakerParams queue.BreakerParams
	Logger        *zap.SugaredLogger
	GetEndpoints  func(RevisionID) (int32, error)
	GetRevision   func(RevisionID) (*v1alpha1.Revision, error)
}

ThrottlerParams defines the parameters of the Throttler.