[VOL-3187]Pass Context down the execution call hierarchy across ofagent codebase
Change-Id: Ia5f2fa1509beefe0ddc427b83e39d2702782db8f
diff --git a/vendor/github.com/uber/jaeger-client-go/internal/throttler/remote/throttler.go b/vendor/github.com/uber/jaeger-client-go/internal/throttler/remote/throttler.go
new file mode 100644
index 0000000..20f434f
--- /dev/null
+++ b/vendor/github.com/uber/jaeger-client-go/internal/throttler/remote/throttler.go
@@ -0,0 +1,216 @@
+// Copyright (c) 2018 The Jaeger Authors.
+//
+// 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.
+
+package remote
+
+import (
+ "fmt"
+ "net/url"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/pkg/errors"
+
+ "github.com/uber/jaeger-client-go"
+ "github.com/uber/jaeger-client-go/utils"
+)
+
+const (
+ // minimumCredits is the minimum amount of credits necessary to not be throttled.
+ // i.e. if currentCredits > minimumCredits, then the operation will not be throttled.
+ minimumCredits = 1.0
+)
+
+var (
+ errorUUIDNotSet = errors.New("Throttler UUID must be set")
+)
+
+type operationBalance struct {
+ Operation string `json:"operation"`
+ Balance float64 `json:"balance"`
+}
+
+type creditResponse struct {
+ Balances []operationBalance `json:"balances"`
+}
+
+type httpCreditManagerProxy struct {
+ hostPort string
+}
+
+func newHTTPCreditManagerProxy(hostPort string) *httpCreditManagerProxy {
+ return &httpCreditManagerProxy{
+ hostPort: hostPort,
+ }
+}
+
+// N.B. Operations list must not be empty.
+func (m *httpCreditManagerProxy) FetchCredits(uuid, serviceName string, operations []string) (*creditResponse, error) {
+ params := url.Values{}
+ params.Set("service", serviceName)
+ params.Set("uuid", uuid)
+ for _, op := range operations {
+ params.Add("operations", op)
+ }
+ var resp creditResponse
+ if err := utils.GetJSON(fmt.Sprintf("http://%s/credits?%s", m.hostPort, params.Encode()), &resp); err != nil {
+ return nil, errors.Wrap(err, "Failed to receive credits from agent")
+ }
+ return &resp, nil
+}
+
+// Throttler retrieves credits from agent and uses it to throttle operations.
+type Throttler struct {
+ options
+
+ mux sync.RWMutex
+ service string
+ uuid atomic.Value
+ creditManager *httpCreditManagerProxy
+ credits map[string]float64 // map of operation->credits
+ close chan struct{}
+ stopped sync.WaitGroup
+}
+
+// NewThrottler returns a Throttler that polls agent for credits and uses them to throttle
+// the service.
+func NewThrottler(service string, options ...Option) *Throttler {
+ opts := applyOptions(options...)
+ creditManager := newHTTPCreditManagerProxy(opts.hostPort)
+ t := &Throttler{
+ options: opts,
+ creditManager: creditManager,
+ service: service,
+ credits: make(map[string]float64),
+ close: make(chan struct{}),
+ }
+ t.stopped.Add(1)
+ go t.pollManager()
+ return t
+}
+
+// IsAllowed implements Throttler#IsAllowed.
+func (t *Throttler) IsAllowed(operation string) bool {
+ t.mux.Lock()
+ defer t.mux.Unlock()
+ value, ok := t.credits[operation]
+ if !ok || value == 0 {
+ if !ok {
+ // NOTE: This appears to be a no-op at first glance, but it stores
+ // the operation key in the map. Necessary for functionality of
+ // Throttler#operations method.
+ t.credits[operation] = 0
+ }
+ if !t.synchronousInitialization {
+ t.metrics.ThrottledDebugSpans.Inc(1)
+ return false
+ }
+ // If it is the first time this operation is being checked, synchronously fetch
+ // the credits.
+ credits, err := t.fetchCredits([]string{operation})
+ if err != nil {
+ // Failed to receive credits from agent, try again next time
+ t.logger.Error("Failed to fetch credits: " + err.Error())
+ return false
+ }
+ if len(credits.Balances) == 0 {
+ // This shouldn't happen but just in case
+ return false
+ }
+ for _, opBalance := range credits.Balances {
+ t.credits[opBalance.Operation] += opBalance.Balance
+ }
+ }
+ return t.isAllowed(operation)
+}
+
+// Close stops the throttler from fetching credits from remote.
+func (t *Throttler) Close() error {
+ close(t.close)
+ t.stopped.Wait()
+ return nil
+}
+
+// SetProcess implements ProcessSetter#SetProcess. It's imperative that the UUID is set before any remote
+// requests are made.
+func (t *Throttler) SetProcess(process jaeger.Process) {
+ if process.UUID != "" {
+ t.uuid.Store(process.UUID)
+ }
+}
+
+// N.B. This function must be called with the Write Lock
+func (t *Throttler) isAllowed(operation string) bool {
+ credits := t.credits[operation]
+ if credits < minimumCredits {
+ t.metrics.ThrottledDebugSpans.Inc(1)
+ return false
+ }
+ t.credits[operation] = credits - minimumCredits
+ return true
+}
+
+func (t *Throttler) pollManager() {
+ defer t.stopped.Done()
+ ticker := time.NewTicker(t.refreshInterval)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ticker.C:
+ t.refreshCredits()
+ case <-t.close:
+ return
+ }
+ }
+}
+
+func (t *Throttler) operations() []string {
+ t.mux.RLock()
+ defer t.mux.RUnlock()
+ operations := make([]string, 0, len(t.credits))
+ for op := range t.credits {
+ operations = append(operations, op)
+ }
+ return operations
+}
+
+func (t *Throttler) refreshCredits() {
+ operations := t.operations()
+ if len(operations) == 0 {
+ return
+ }
+ newCredits, err := t.fetchCredits(operations)
+ if err != nil {
+ t.metrics.ThrottlerUpdateFailure.Inc(1)
+ t.logger.Error("Failed to fetch credits: " + err.Error())
+ return
+ }
+ t.metrics.ThrottlerUpdateSuccess.Inc(1)
+
+ t.mux.Lock()
+ defer t.mux.Unlock()
+ for _, opBalance := range newCredits.Balances {
+ t.credits[opBalance.Operation] += opBalance.Balance
+ }
+}
+
+func (t *Throttler) fetchCredits(operations []string) (*creditResponse, error) {
+ uuid := t.uuid.Load()
+ uuidStr, _ := uuid.(string)
+ if uuid == nil || uuidStr == "" {
+ return nil, errorUUIDNotSet
+ }
+ return t.creditManager.FetchCredits(uuidStr, t.service, operations)
+}