123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358 |
- // Copyright (c) 2017 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.
- package jaeger
- import (
- "encoding/json"
- "fmt"
- "io/ioutil"
- "net/http"
- "net/url"
- "sync"
- "sync/atomic"
- "time"
- "github.com/uber/jaeger-client-go/log"
- "github.com/uber/jaeger-client-go/thrift-gen/sampling"
- )
- const (
- defaultRemoteSamplingTimeout = 10 * time.Second
- defaultSamplingRefreshInterval = time.Minute
- )
- // SamplingStrategyFetcher is used to fetch sampling strategy updates from remote server.
- type SamplingStrategyFetcher interface {
- Fetch(service string) ([]byte, error)
- }
- // SamplingStrategyParser is used to parse sampling strategy updates. The output object
- // should be of the type that is recognized by the SamplerUpdaters.
- type SamplingStrategyParser interface {
- Parse(response []byte) (interface{}, error)
- }
- // SamplerUpdater is used by RemotelyControlledSampler to apply sampling strategies,
- // retrieved from remote config server, to the current sampler. The updater can modify
- // the sampler in-place if sampler supports it, or create a new one.
- //
- // If the strategy does not contain configuration for the sampler in question,
- // updater must return modifiedSampler=nil to give other updaters a chance to inspect
- // the sampling strategy response.
- //
- // RemotelyControlledSampler invokes the updaters while holding a lock on the main sampler.
- type SamplerUpdater interface {
- Update(sampler SamplerV2, strategy interface{}) (modified SamplerV2, err error)
- }
- // RemotelyControlledSampler is a delegating sampler that polls a remote server
- // for the appropriate sampling strategy, constructs a corresponding sampler and
- // delegates to it for sampling decisions.
- type RemotelyControlledSampler struct {
- // These fields must be first in the struct because `sync/atomic` expects 64-bit alignment.
- // Cf. https://github.com/uber/jaeger-client-go/issues/155, https://goo.gl/zW7dgq
- closed int64 // 0 - not closed, 1 - closed
- sync.RWMutex // used to serialize access to samplerOptions.sampler
- samplerOptions
- serviceName string
- doneChan chan *sync.WaitGroup
- }
- // NewRemotelyControlledSampler creates a sampler that periodically pulls
- // the sampling strategy from an HTTP sampling server (e.g. jaeger-agent).
- func NewRemotelyControlledSampler(
- serviceName string,
- opts ...SamplerOption,
- ) *RemotelyControlledSampler {
- options := new(samplerOptions).applyOptionsAndDefaults(opts...)
- sampler := &RemotelyControlledSampler{
- samplerOptions: *options,
- serviceName: serviceName,
- doneChan: make(chan *sync.WaitGroup),
- }
- go sampler.pollController()
- return sampler
- }
- // IsSampled implements IsSampled() of Sampler.
- // TODO (breaking change) remove when Sampler V1 is removed
- func (s *RemotelyControlledSampler) IsSampled(id TraceID, operation string) (bool, []Tag) {
- return false, nil
- }
- // OnCreateSpan implements OnCreateSpan of SamplerV2.
- func (s *RemotelyControlledSampler) OnCreateSpan(span *Span) SamplingDecision {
- s.RLock()
- defer s.RUnlock()
- return s.sampler.OnCreateSpan(span)
- }
- // OnSetOperationName implements OnSetOperationName of SamplerV2.
- func (s *RemotelyControlledSampler) OnSetOperationName(span *Span, operationName string) SamplingDecision {
- s.RLock()
- defer s.RUnlock()
- return s.sampler.OnSetOperationName(span, operationName)
- }
- // OnSetTag implements OnSetTag of SamplerV2.
- func (s *RemotelyControlledSampler) OnSetTag(span *Span, key string, value interface{}) SamplingDecision {
- s.RLock()
- defer s.RUnlock()
- return s.sampler.OnSetTag(span, key, value)
- }
- // OnFinishSpan implements OnFinishSpan of SamplerV2.
- func (s *RemotelyControlledSampler) OnFinishSpan(span *Span) SamplingDecision {
- s.RLock()
- defer s.RUnlock()
- return s.sampler.OnFinishSpan(span)
- }
- // Close implements Close() of Sampler.
- func (s *RemotelyControlledSampler) Close() {
- if swapped := atomic.CompareAndSwapInt64(&s.closed, 0, 1); !swapped {
- s.logger.Error("Repeated attempt to close the sampler is ignored")
- return
- }
- var wg sync.WaitGroup
- wg.Add(1)
- s.doneChan <- &wg
- wg.Wait()
- }
- // Equal implements Equal() of Sampler.
- func (s *RemotelyControlledSampler) Equal(other Sampler) bool {
- // NB The Equal() function is expensive and will be removed. See PerOperationSampler.Equal() for
- // more information.
- return false
- }
- func (s *RemotelyControlledSampler) pollController() {
- ticker := time.NewTicker(s.samplingRefreshInterval)
- defer ticker.Stop()
- s.pollControllerWithTicker(ticker)
- }
- func (s *RemotelyControlledSampler) pollControllerWithTicker(ticker *time.Ticker) {
- for {
- select {
- case <-ticker.C:
- s.UpdateSampler()
- case wg := <-s.doneChan:
- wg.Done()
- return
- }
- }
- }
- // Sampler returns the currently active sampler.
- func (s *RemotelyControlledSampler) Sampler() SamplerV2 {
- s.RLock()
- defer s.RUnlock()
- return s.sampler
- }
- func (s *RemotelyControlledSampler) setSampler(sampler SamplerV2) {
- s.Lock()
- defer s.Unlock()
- s.sampler = sampler
- }
- // UpdateSampler forces the sampler to fetch sampling strategy from backend server.
- // This function is called automatically on a timer, but can also be safely called manually, e.g. from tests.
- func (s *RemotelyControlledSampler) UpdateSampler() {
- res, err := s.samplingFetcher.Fetch(s.serviceName)
- if err != nil {
- s.metrics.SamplerQueryFailure.Inc(1)
- s.logger.Infof("failed to fetch sampling strategy: %v", err)
- return
- }
- strategy, err := s.samplingParser.Parse(res)
- if err != nil {
- s.metrics.SamplerUpdateFailure.Inc(1)
- s.logger.Infof("failed to parse sampling strategy response: %v", err)
- return
- }
- s.Lock()
- defer s.Unlock()
- s.metrics.SamplerRetrieved.Inc(1)
- if err := s.updateSamplerViaUpdaters(strategy); err != nil {
- s.metrics.SamplerUpdateFailure.Inc(1)
- s.logger.Infof("failed to handle sampling strategy response %+v. Got error: %v", res, err)
- return
- }
- s.metrics.SamplerUpdated.Inc(1)
- }
- // NB: this function should only be called while holding a Write lock
- func (s *RemotelyControlledSampler) updateSamplerViaUpdaters(strategy interface{}) error {
- for _, updater := range s.updaters {
- sampler, err := updater.Update(s.sampler, strategy)
- if err != nil {
- return err
- }
- if sampler != nil {
- s.logger.Debugf("sampler updated: %+v", sampler)
- s.sampler = sampler
- return nil
- }
- }
- return fmt.Errorf("unsupported sampling strategy %+v", strategy)
- }
- // -----------------------
- // ProbabilisticSamplerUpdater is used by RemotelyControlledSampler to parse sampling configuration.
- type ProbabilisticSamplerUpdater struct{}
- // Update implements Update of SamplerUpdater.
- func (u *ProbabilisticSamplerUpdater) Update(sampler SamplerV2, strategy interface{}) (SamplerV2, error) {
- type response interface {
- GetProbabilisticSampling() *sampling.ProbabilisticSamplingStrategy
- }
- var _ response = new(sampling.SamplingStrategyResponse) // sanity signature check
- if resp, ok := strategy.(response); ok {
- if probabilistic := resp.GetProbabilisticSampling(); probabilistic != nil {
- if ps, ok := sampler.(*ProbabilisticSampler); ok {
- if err := ps.Update(probabilistic.SamplingRate); err != nil {
- return nil, err
- }
- return sampler, nil
- }
- return newProbabilisticSampler(probabilistic.SamplingRate), nil
- }
- }
- return nil, nil
- }
- // -----------------------
- // RateLimitingSamplerUpdater is used by RemotelyControlledSampler to parse sampling configuration.
- type RateLimitingSamplerUpdater struct{}
- // Update implements Update of SamplerUpdater.
- func (u *RateLimitingSamplerUpdater) Update(sampler SamplerV2, strategy interface{}) (SamplerV2, error) {
- type response interface {
- GetRateLimitingSampling() *sampling.RateLimitingSamplingStrategy
- }
- var _ response = new(sampling.SamplingStrategyResponse) // sanity signature check
- if resp, ok := strategy.(response); ok {
- if rateLimiting := resp.GetRateLimitingSampling(); rateLimiting != nil {
- rateLimit := float64(rateLimiting.MaxTracesPerSecond)
- if rl, ok := sampler.(*RateLimitingSampler); ok {
- rl.Update(rateLimit)
- return rl, nil
- }
- return NewRateLimitingSampler(rateLimit), nil
- }
- }
- return nil, nil
- }
- // -----------------------
- // AdaptiveSamplerUpdater is used by RemotelyControlledSampler to parse sampling configuration.
- // Fields have the same meaning as in PerOperationSamplerParams.
- type AdaptiveSamplerUpdater struct {
- MaxOperations int
- OperationNameLateBinding bool
- }
- // Update implements Update of SamplerUpdater.
- func (u *AdaptiveSamplerUpdater) Update(sampler SamplerV2, strategy interface{}) (SamplerV2, error) {
- type response interface {
- GetOperationSampling() *sampling.PerOperationSamplingStrategies
- }
- var _ response = new(sampling.SamplingStrategyResponse) // sanity signature check
- if p, ok := strategy.(response); ok {
- if operations := p.GetOperationSampling(); operations != nil {
- if as, ok := sampler.(*PerOperationSampler); ok {
- as.update(operations)
- return as, nil
- }
- return NewPerOperationSampler(PerOperationSamplerParams{
- MaxOperations: u.MaxOperations,
- OperationNameLateBinding: u.OperationNameLateBinding,
- Strategies: operations,
- }), nil
- }
- }
- return nil, nil
- }
- // -----------------------
- type httpSamplingStrategyFetcher struct {
- serverURL string
- logger log.DebugLogger
- httpClient http.Client
- }
- func newHTTPSamplingStrategyFetcher(serverURL string, logger log.DebugLogger) *httpSamplingStrategyFetcher {
- customTransport := http.DefaultTransport.(*http.Transport).Clone()
- customTransport.ResponseHeaderTimeout = defaultRemoteSamplingTimeout
- return &httpSamplingStrategyFetcher{
- serverURL: serverURL,
- logger: logger,
- httpClient: http.Client{
- Transport: customTransport,
- },
- }
- }
- func (f *httpSamplingStrategyFetcher) Fetch(serviceName string) ([]byte, error) {
- v := url.Values{}
- v.Set("service", serviceName)
- uri := f.serverURL + "?" + v.Encode()
- resp, err := f.httpClient.Get(uri)
- if err != nil {
- return nil, err
- }
- defer func() {
- if err := resp.Body.Close(); err != nil {
- f.logger.Error(fmt.Sprintf("failed to close HTTP response body: %+v", err))
- }
- }()
- body, err := ioutil.ReadAll(resp.Body)
- if err != nil {
- return nil, err
- }
- if resp.StatusCode >= 400 {
- return nil, fmt.Errorf("StatusCode: %d, Body: %s", resp.StatusCode, body)
- }
- return body, nil
- }
- // -----------------------
- type samplingStrategyParser struct{}
- func (p *samplingStrategyParser) Parse(response []byte) (interface{}, error) {
- strategy := new(sampling.SamplingStrategyResponse)
- if err := json.Unmarshal(response, strategy); err != nil {
- return nil, err
- }
- return strategy, nil
- }
|