dedup

Documentation

Overview

Copyright (c) 2016-2019 Uber Technologies, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright (c) 2016-2019 Uber Technologies, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Copyright (c) 2016-2019 Uber Technologies, Inc.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• Constants
• Variables
• type ErrorMatcher
• type IntervalTask
• type IntervalTrap
  • func NewIntervalTrap(interval time.Duration, clk clock.Clock, task IntervalTask) *IntervalTrap
  • func (t *IntervalTrap) Trap()
• type Limiter
  • func NewLimiter(clk clock.Clock, runner TaskRunner) *Limiter
  • func (l *Limiter) Run(input interface{}) interface{}
• type Request
• type RequestCache
  • func NewRequestCache(config RequestCacheConfig, clk clock.Clock) *RequestCache
  • func (c *RequestCache) SetNotFound(m ErrorMatcher)
  • func (c *RequestCache) Start(id string, r Request) error
• type RequestCacheConfig
• type TaskRunner

Constants

const TaskGCInterval = time.Minute

TaskGCInterval is the interval in which garbage collection of old tasks runs.

Variables

var (
	ErrRequestPending = errors.New("request pending")
	ErrWorkersBusy    = errors.New("no workers available to handle request")
)

RequestCache errors.

Types

type ErrorMatcher

type ErrorMatcher func(error) bool

ErrorMatcher defines functions which RequestCache uses to detect user defined errors.

type IntervalTask

type IntervalTask interface {
	Run()
}

IntervalTask defines a task to run on some interval.

type IntervalTrap

type IntervalTrap struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

IntervalTrap manages trapping into some task which must run at a given interval.

func NewIntervalTrap

func NewIntervalTrap(
	interval time.Duration, clk clock.Clock, task IntervalTask) *IntervalTrap

NewIntervalTrap creates a new IntervalTrap.

func (*IntervalTrap) Trap

func (t *IntervalTrap) Trap()

Trap quickly checks if the interval has passed since the last task run, and if it has, it runs the task.

type Limiter

type Limiter struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Limiter deduplicates the running of a common task within a given limit. Tasks are deduplicated based on input equality.

func NewLimiter

func NewLimiter(clk clock.Clock, runner TaskRunner) *Limiter

NewLimiter creates a new Limiter for tasks. The limit is determined per task via the TaskRunner.

func (*Limiter) Run

func (l *Limiter) Run(input interface{}) interface{}

Run runs a task with input.

type Request

type Request func() error

Request defines functions which encapsulate a request.

type RequestCache

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

RequestCache tracks pending requests and caches errors for configurable TTLs. It is used to prevent request duplication and DDOS-ing external components. Each request is represented by an arbitrary id string determined by the user.

func NewRequestCache

func NewRequestCache(config RequestCacheConfig, clk clock.Clock) *RequestCache

NewRequestCache creates a new RequestCache.

func (*RequestCache) SetNotFound

func (c *RequestCache) SetNotFound(m ErrorMatcher)

SetNotFound sets the ErrorMatcher for activating the configured NotFoundTTL for errors returned by Request functions.

func (*RequestCache) Start

func (c *RequestCache) Start(id string, r Request) error

Start concurrently runs r under the given id. Any error returned by r will be cached for the configured TTL. If there is already a function executing under id, Start returns ErrRequestPending. If there are no available workers to run r, Start returns ErrWorkersBusy.

type RequestCacheConfig

type RequestCacheConfig struct {
	NotFoundTTL     time.Duration `yaml:"not_found_ttl"`
	ErrorTTL        time.Duration `yaml:"error_ttl"`
	CleanupInterval time.Duration `yaml:"cleanup_interval"`
	NumWorkers      int           `yaml:"num_workers"`
	BusyTimeout     time.Duration `yaml:"busy_timeout"`
}

RequestCacheConfig defines RequestCache configuration.

type TaskRunner

type TaskRunner interface {
	Run(input interface{}) (output interface{}, ttl time.Duration)
}

TaskRunner runs against some input and produces some output w/ a ttl.