public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tracing: Support configuring Jaeger client from environment (#21103)

Browse Source 
* Tracing: Support configuring Jaeger client from environment
* Tracing: Replace deprecated Jaeger Client call
This commit is contained in:
Dave Henderson
2019-12-18 20:09:15 +01:00
committed by Arve Knudsen
parent 7aeba652c9
commit 66b70b50b8
73 changed files with 11242 additions and 724 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vendor/github.com/uber/jaeger-client-go/CHANGELOG.md
Generated Vendored
+80
View File
@@ -1,6 +1,86 @@
Changes by Version
==================
2.20.1 (2019-11-08)
-------------------
Minor patch via https://github.com/jaegertracing/jaeger-client-go/pull/468
- Make `AdaptiveSamplerUpdater` usable with default values; Resolves #467
- Create `OperationNameLateBinding` sampler option and config option
- Make `SamplerOptions` var of public type, so that its functions are discoverable via godoc
2.20.0 (2019-11-06)
-------------------
## New Features
- Allow all in-process spans of a trace to share sampling state (#443) -- Prithvi Raj
Sampling state is shared between all spans of the trace that are still in memory.
This allows implementation of delayed sampling decisions (see below).
- Support delayed sampling decisions (#449) -- Yuri Shkuro
This is a large structural change to how the samplers work.
It allows some samplers to be executed multiple times on different
span events (like setting a tag) and make a positive sampling decision
later in the span life cycle, or even based on children spans.
See [README](./README.md#delayed-sampling) for more details.
There is a related minor change in behavior of the adaptive (per-operation) sampler,
which will no longer re-sample the trace when `span.SetOperation()` is called, i.e. the
operation used to make the sampling decision is always the one provided at span creation.
- Add experimental tag matching sampler (#452) -- Yuri Shkuro
A sampler that can sample a trace based on a certain tag added to the root
span or one of its local (in-process) children. The sampler can be used with
another experimental `PrioritySampler` that allows multiple samplers to try
to make a sampling decision, in a certain priority order.
- [log/zap] Report whether a trace was sampled (#445) -- Abhinav Gupta
- Allow config.FromEnv() to enrich an existing config object (#436) -- Vineeth Reddy
## Minor patches
- Expose Sampler on Tracer and accept sampler options via Configuration (#460) -- Yuri Shkuro
- Fix github.com/uber-go/atomic import (#464) -- Yuri Shkuro
- Add nodejs to crossdock tests (#441) -- Bhavin Gandhi
- Bump Go compiler version to 1.13 (#453) -- Yuri Shkuro
2.19.0 (2019-09-23)
-------------------
- Upgrade jaeger-lib to 2.2 and unpin Prom client (#434) -- Yuri Shkuro
2.18.1 (2019-09-16)
-------------------
- Remove go.mod / go.sum that interfere with `go get` (#432)
2.18.0 (2019-09-09)
-------------------
- Add option "noDebugFlagOnForcedSampling" for tracer initialization [resolves #422] (#423) <Jun Guo>
2.17.0 (2019-08-30)
-------------------
- Add a flag for firehose mode (#419) <Prithvi Raj>
- Default sampling server URL to agent (#414) <Bryan Boreham>
- Update default sampling rate when sampling strategy is refreshed (#413) <Bryan Boreham>
- Support "Self" Span Reference (#411) <dm03514>
- Don't complain about blank service name if tracing is Disabled (#410) Yuri <Shkuro>
- Use IP address from tag if exist (#402) <NikoKVCS>
- Expose span data to custom reporters [fixes #394] (#399) <Curtis Allen>
- Fix the span allocation in the pool (#381) <Dmitry Ponomarev>
2.16.0 (2019-03-24)
-------------------
vendor/github.com/uber/jaeger-client-go/CONTRIBUTING.md
Generated Vendored
+5 -5
View File
@@ -19,7 +19,7 @@ file for details.
## Getting Started
This library uses [glide](https://github.com/Masterminds/glide) to manage dependencies.
This library uses [dep](https://golang.github.io/dep/) to manage dependencies.
To get started, make sure you clone the Git repository into the correct location
`github.com/uber/jaeger-client-go` relative to `$GOPATH`:
@@ -29,13 +29,13 @@ mkdir -p $GOPATH/src/github.com/uber
cd $GOPATH/src/github.com/uber
git clone git@github.com:jaegertracing/jaeger-client-go.git jaeger-client-go
cd jaeger-client-go
git submodule update --init --recursive
```
Then install dependencies and run the tests:
```
git submodule update --init --recursive
glide install
make install
make test
```
@@ -45,13 +45,13 @@ This projects follows the following pattern for grouping imports in Go files:
* imports from standard library
* imports from other projects
* imports from `jaeger-client-go` project
For example:
```go
import (
"fmt"
"github.com/uber/jaeger-lib/metrics"
"go.uber.org/zap"
vendor/github.com/uber/jaeger-client-go/Gopkg.lock
Generated Vendored
+138 -44
View File
@@ -2,12 +2,20 @@
[[projects]]
branch = "master"
digest = "1:9f3b30d9f8e0d7040f729b82dcbc8f0dead820a133b3147ce355fc451f32d761"
name = "github.com/BurntSushi/toml"
packages = ["."]
pruneopts = "UT"
revision = "3012a1dbe2e4bd1391d42b32f0577cb7bbc7f005"
version = "v0.3.1"
[[projects]]
digest = "1:d6afaeed1502aa28e80a4ed0981d570ad91b2579193404256ce672ed0a609e0d"
name = "github.com/beorn7/perks"
packages = ["quantile"]
pruneopts = "UT"
revision = "3a771d992973f24aa725d07868b467d1ddfceafb"
revision = "37c8de3658fcb183f997c4e13e8337516ab753e6"
version = "v1.0.1"
[[projects]]
branch = "master"
@@ -38,12 +46,12 @@
version = "v1.1.1"
[[projects]]
digest = "1:318f1c959a8a740366fce4b1e1eb2fd914036b4af58fbd0a003349b305f118ad"
digest = "1:573ca21d3669500ff845bdebee890eb7fc7f0f50c59f2132f2a0c6b03d85086a"
name = "github.com/golang/protobuf"
packages = ["proto"]
pruneopts = "UT"
revision = "b5d812f8a3706043e23a9cd5babf2e5423744d30"
version = "v1.3.1"
revision = "6c65a5562fc06764971b7c5d05c76c75e84bdbf7"
version = "v1.3.2"
[[projects]]
digest = "1:ff5ebae34cfbf047d505ee150de27e60570e8c394b3b8fdbb720ff6ac71985fc"
@@ -83,12 +91,15 @@
version = "v1.0.0"
[[projects]]
digest = "1:b6221ec0f8903b556e127c449e7106b63e6867170c2d10a7c058623d086f2081"
digest = "1:7097829edd12fd7211fca0d29496b44f94ef9e6d72f88fb64f3d7b06315818ad"
name = "github.com/prometheus/client_golang"
packages = ["prometheus"]
packages = [
"prometheus",
"prometheus/internal",
]
pruneopts = "UT"
revision = "c5b7fccd204277076155f10851dad72b76a49317"
version = "v0.8.0"
revision = "170205fb58decfd011f1550d4cfb737230d7ae4f"
version = "v1.1.0"
[[projects]]
branch = "master"
@@ -96,10 +107,10 @@
name = "github.com/prometheus/client_model"
packages = ["go"]
pruneopts = "UT"
revision = "fd36f4220a901265f90734c3183c5f0c91daa0b8"
revision = "14fe0d1b01d4d5fc031dd4bec1823bd3ebbe8016"
[[projects]]
digest = "1:35cf6bdf68db765988baa9c4f10cc5d7dda1126a54bd62e252dbcd0b1fc8da90"
digest = "1:f119e3205d3a1f0f19dbd7038eb37528e2c6f0933269dc344e305951fb87d632"
name = "github.com/prometheus/common"
packages = [
"expfmt",
@@ -107,25 +118,23 @@
"model",
]
pruneopts = "UT"
revision = "cfeb6f9992ffa54aaa4f2170ade4067ee478b250"
version = "v0.2.0"
revision = "287d3e634a1e550c9e463dd7e5a75a422c614505"
version = "v0.7.0"
[[projects]]
branch = "master"
digest = "1:c31163bd62461e0c5f7ddc7363e39ef8d9e929693e77b5c11c709b05f9cb9219"
digest = "1:a210815b437763623ecca8eb91e6a0bf4f2d6773c5a6c9aec0e28f19e5fd6deb"
name = "github.com/prometheus/procfs"
packages = [
".",
"internal/fs",
"internal/util",
"iostats",
"nfs",
"xfs",
]
pruneopts = "UT"
revision = "55ae3d9d557340b5bc24cd8aa5f6fa2c2ab31352"
revision = "499c85531f756d1129edd26485a5f73871eeb308"
version = "v0.0.5"
[[projects]]
digest = "1:8ff03ccc603abb0d7cce94d34b613f5f6251a9e1931eba1a3f9888a9029b055c"
digest = "1:0496f0e99014b7fd0a560c539f51d0882731137b85494142f47e550e4657176a"
name = "github.com/stretchr/testify"
packages = [
"assert",
@@ -133,19 +142,11 @@
"suite",
]
pruneopts = "UT"
revision = "ffdc059bfe9ce6a4e144ba849dbedead332c6053"
version = "v1.3.0"
revision = "221dbe5ed46703ee255b1da0dec05086f5035f62"
version = "v1.4.0"
[[projects]]
digest = "1:3c1a69cdae3501bf75e76d0d86dc6f2b0a7421bc205c0cb7b96b19eed464a34d"
name = "github.com/uber-go/atomic"
packages = ["."]
pruneopts = "UT"
revision = "1ea20fb1cbb1cc08cbd0d913a96dead89aa18289"
version = "v1.3.2"
[[projects]]
digest = "1:f5c5ad1e08141e18aee1b9c37729d93d06805840421ccfc9d407787ffe969ce6"
digest = "1:0ec60ffd594af00ba1660bc746aa0e443d27dd4003dee55f9d08a0b4ff5431a3"
name = "github.com/uber/jaeger-lib"
packages = [
"metrics",
@@ -153,27 +154,35 @@
"metrics/prometheus",
]
pruneopts = "UT"
revision = "0e30338a695636fe5bcf7301e8030ce8dd2a8530"
version = "v2.0.0"
revision = "a87ae9d84fb038a8d79266298970720be7c80fcd"
version = "v2.2.0"
[[projects]]
digest = "1:3c1a69cdae3501bf75e76d0d86dc6f2b0a7421bc205c0cb7b96b19eed464a34d"
digest = "1:0bdcb0c740d79d400bd3f7946ac22a715c94db62b20bfd2e01cd50693aba0600"
name = "go.uber.org/atomic"
packages = ["."]
pruneopts = "UT"
revision = "1ea20fb1cbb1cc08cbd0d913a96dead89aa18289"
version = "v1.3.2"
revision = "9dc4df04d0d1c39369750a9f6c32c39560672089"
version = "v1.5.0"
[[projects]]
digest = "1:60bf2a5e347af463c42ed31a493d817f8a72f102543060ed992754e689805d1a"
digest = "1:002ebc50f3ef475ac325e1904be931d9dcba6dc6d73b5682afce0c63436e3902"
name = "go.uber.org/multierr"
packages = ["."]
pruneopts = "UT"
revision = "3c4937480c32f4c13a875a1829af76c98ca3d40a"
version = "v1.1.0"
revision = "c3fc3d02ec864719d8e25be2d7dde1e35a36aa27"
version = "v1.3.0"
[[projects]]
digest = "1:c52caf7bd44f92e54627a31b85baf06a68333a196b3d8d241480a774733dcf8b"
branch = "master"
digest = "1:3032e90a153750ea149f68bf081f97ca738f041fba45c41c80737f572ffdf2f4"
name = "go.uber.org/tools"
packages = ["update-license"]
pruneopts = "UT"
revision = "2cfd321de3ee5d5f8a5fda2521d1703478334d98"
[[projects]]
digest = "1:6be13632ab4bd5842a097abb3aabac045a8601e19a10da4239e7d8bd83d4b83c"
name = "go.uber.org/zap"
packages = [
".",
@@ -184,8 +193,19 @@
"zapcore",
]
pruneopts = "UT"
revision = "ff33455a0e382e8a81d14dd7c922020b6b5e7982"
version = "v1.9.1"
revision = "a6015e13fab9b744d96085308ce4e8f11bad1996"
version = "v1.12.0"
[[projects]]
branch = "master"
digest = "1:21d7bad9b7da270fd2d50aba8971a041bd691165c95096a2a4c68db823cbc86a"
name = "golang.org/x/lint"
packages = [
".",
"golint",
]
pruneopts = "UT"
revision = "16217165b5de779cb6a5e4fc81fa9c1166fda457"
[[projects]]
branch = "master"
@@ -196,7 +216,81 @@
"context/ctxhttp",
]
pruneopts = "UT"
revision = "addf6b3196f61cd44ce5a76657913698c73479d0"
revision = "0deb6923b6d97481cb43bc1043fe5b72a0143032"
[[projects]]
branch = "master"
digest = "1:5dfb17d45415b7b8927382f53955a66f55f9d9d11557aa82f7f481d642ab247a"
name = "golang.org/x/sys"
packages = ["windows"]
pruneopts = "UT"
revision = "f43be2a4598cf3a47be9f94f0c28197ed9eae611"
[[projects]]
branch = "master"
digest = "1:bae8b3bf837d9d7f601776f37f44e031d46943677beff8fb2eb9c7317d44de2f"
name = "golang.org/x/tools"
packages = [
"go/analysis",
"go/analysis/passes/inspect",
"go/ast/astutil",
"go/ast/inspector",
"go/buildutil",
"go/gcexportdata",
"go/internal/gcimporter",
"go/internal/packagesdriver",
"go/packages",
"go/types/objectpath",
"go/types/typeutil",
"internal/fastwalk",
"internal/gopathwalk",
"internal/semver",
"internal/span",
]
pruneopts = "UT"
revision = "8dbcdeb83d3faec5315146800b375c4962a42fc6"
[[projects]]
digest = "1:59f10c1537d2199d9115d946927fe31165959a95190849c82ff11e05803528b0"
name = "gopkg.in/yaml.v2"
packages = ["."]
pruneopts = "UT"
revision = "f221b8435cfb71e54062f6c6e99e9ade30b124d5"
version = "v2.2.4"
[[projects]]
digest = "1:131158a88aad1f94854d0aa21a64af2802d0a470fb0f01cb33c04fafd2047111"
name = "honnef.co/go/tools"
packages = [
"arg",
"cmd/staticcheck",
"config",
"deprecated",
"facts",
"functions",
"go/types/typeutil",
"internal/cache",
"internal/passes/buildssa",
"internal/renameio",
"internal/sharedcheck",
"lint",
"lint/lintdsl",
"lint/lintutil",
"lint/lintutil/format",
"loader",
"printf",
"simple",
"ssa",
"ssautil",
"staticcheck",
"staticcheck/vrp",
"stylecheck",
"unused",
"version",
]
pruneopts = "UT"
revision = "afd67930eec2a9ed3e9b19f684d17a062285f16a"
version = "2019.2.3"
[solve-meta]
analyzer-name = "dep"
@@ -212,10 +306,10 @@
"github.com/stretchr/testify/assert",
"github.com/stretchr/testify/require",
"github.com/stretchr/testify/suite",
"github.com/uber-go/atomic",
"github.com/uber/jaeger-lib/metrics",
"github.com/uber/jaeger-lib/metrics/metricstest",
"github.com/uber/jaeger-lib/metrics/prometheus",
"go.uber.org/atomic",
"go.uber.org/zap",
"go.uber.org/zap/zapcore",
]
vendor/github.com/uber/jaeger-client-go/Gopkg.toml
Generated Vendored
+3 -3
View File
@@ -8,19 +8,19 @@
[[constraint]]
name = "github.com/prometheus/client_golang"
version = "0.8.0"
version = "^1"
[[constraint]]
name = "github.com/stretchr/testify"
version = "^1.1.3"
[[constraint]]
name = "github.com/uber-go/atomic"
name = "go.uber.org/atomic"
version = "^1"
[[constraint]]
name = "github.com/uber/jaeger-lib"
version = "^2.0"
version = "^2.2"
[[constraint]]
name = "go.uber.org/zap"
vendor/github.com/uber/jaeger-client-go/Makefile
Generated Vendored
+17 -12
View File
@@ -1,14 +1,14 @@
PROJECT_ROOT=github.com/uber/jaeger-client-go
PACKAGES := $(shell glide novendor | grep -v -e ./thrift-gen/... -e ./thrift/...)
PACKAGES := . $(shell go list ./... | awk -F/ 'NR>1 {print "./"$$4"/..."}' | grep -v -e ./thrift-gen/... -e ./thrift/... | sort -u)
# all .go files that don't exist in hidden directories
ALL_SRC := $(shell find . -name "*.go" | grep -v -e vendor -e thrift-gen -e ./thrift/ \
-e ".*/\..*" \
-e ".*/_.*" \
-e ".*/mocks.*")
-include crossdock/rules.mk
USE_DEP := true
export GO15VENDOREXPERIMENT=1
-include crossdock/rules.mk
RACE=-race
GOTEST=go test -v $(RACE)
@@ -58,19 +58,24 @@ lint:
.PHONY: install
install:
glide --version || go get github.com/Masterminds/glide
@echo install: USE_DEP=$(USE_DEP) USE_GLIDE=$(USE_GLIDE)
ifeq ($(USE_DEP),true)
dep version || make install-dep
dep ensure
else
endif
ifeq ($(USE_GLIDE),true)
glide --version || go get github.com/Masterminds/glide
glide install
endif
.PHONY: cover
cover:
./scripts/cover.sh $(shell go list $(PACKAGES))
go tool cover -html=cover.out -o cover.html
$(GOTEST) -cover -coverprofile cover.out $(PACKAGES)
.PHONY: cover-html
cover-html: cover
go tool cover -html=cover.out -o cover.html
# This is not part of the regular test target because we don't want to slow it
# down.
@@ -101,23 +106,23 @@ idl-submodule:
thrift-image:
$(THRIFT) -version
.PHONY: install-dep-ci
install-dep-ci:
.PHONY: install-dep
install-dep:
- curl -L -s https://github.com/golang/dep/releases/download/v0.5.0/dep-linux-amd64 -o $$GOPATH/bin/dep
- chmod +x $$GOPATH/bin/dep
.PHONY: install-ci
install-ci: install-dep-ci install
install-ci: install
go get github.com/wadey/gocovmerge
go get github.com/mattn/goveralls
go get golang.org/x/tools/cmd/cover
go get golang.org/x/lint/golint
.PHONY: test-ci
test-ci:
@./scripts/cover.sh $(shell go list $(PACKAGES))
test-ci: cover
ifeq ($(CI_SKIP_LINT),true)
echo 'skipping lint'
else
make lint
endif
vendor/github.com/uber/jaeger-client-go/README.md
Generated Vendored
+49 -7
View File
@@ -3,7 +3,7 @@
# Jaeger Bindings for Go OpenTracing API
Instrumentation library that implements an
[OpenTracing](http://opentracing.io) Tracer for Jaeger (https://jaegertracing.io).
[OpenTracing Go](https://github.com/opentracing/opentracing-go) Tracer for Jaeger (https://jaegertracing.io).
**IMPORTANT**: The library's import path is based on its original location under `github.com/uber`. Do not try to import it as `github.com/jaegertracing`, it will not compile. We might revisit this in the next major release.
* :white_check_mark: `import "github.com/uber/jaeger-client-go"`
@@ -15,19 +15,20 @@ Please see [CONTRIBUTING.md](CONTRIBUTING.md).
## Installation
We recommended using a dependency manager like [glide](https://github.com/Masterminds/glide)
We recommended using a dependency manager like [dep](https://golang.github.io/dep/)
and [semantic versioning](http://semver.org/) when including this library into an application.
For example, Jaeger backend imports this library like this:
```yaml
- package: github.com/uber/jaeger-client-go
version: ^2.7.0
```toml
[[constraint]]
name = "github.com/uber/jaeger-client-go"
version = "2.17"
```
If you instead want to use the latest version in `master`, you can pull it via `go get`.
Note that during `go get` you may see build errors due to incompatible dependencies, which is why
we recommend using semantic versions for dependencies. The error may be fixed by running
`make install` (it will install `glide` if you don't have it):
`make install` (it will install `dep` if you don't have it):
```shell
go get -u github.com/uber/jaeger-client-go/
@@ -181,6 +182,29 @@ are available:
1. `RateLimitingSampler` can be used to allow only a certain fixed
number of traces to be sampled per second.
#### Delayed sampling
Version 2.20 introduced the ability to delay sampling decisions in the life cycle
of the root span. It involves several features and architectural changes:
* **Shared sampling state**: the sampling state is shared across all local
(i.e. in-process) spans for a given trace.
* **New `SamplerV2` API** allows the sampler to be called at multiple points
in the life cycle of a span:
* on span creation
* on overwriting span operation name
* on setting span tags
* on finishing the span
* **Final/non-final sampling state**: the new `SamplerV2` API allows the sampler
to indicate if the negative sampling decision is final or not (positive sampling
decisions are always final). If the decision is not final, the sampler will be
called again on further span life cycle events, like setting tags.
These new features are used in the experimental `x.TagMatchingSampler`, which
can sample a trace based on a certain tag added to the root
span or one of its local (in-process) children. The sampler can be used with
another experimental `x.PrioritySampler` that allows multiple samplers to try
to make a sampling decision, in a certain priority order.
### Baggage Injection
The OpenTracing spec allows for [baggage][baggage], which are key value pairs that are added
@@ -222,7 +246,7 @@ import (
)
span := opentracing.SpanFromContext(ctx)
ext.SamplingPriority.Set(span, 1)
ext.SamplingPriority.Set(span, 1)
```
#### Via HTTP Headers
@@ -253,6 +277,24 @@ by a lot of Zipkin tracers. This means that you can use Jaeger in conjunction wi
However it is not the default propagation format, see [here](zipkin/README.md#NewZipkinB3HTTPHeaderPropagator) how to set it up.
## SelfRef
Jaeger Tracer supports an additional [reference](https://github.com/opentracing/specification/blob/1.1/specification.md#references-between-spans)
type call `Self`. This allows a caller to provide an already established `SpanContext`.
This allows loading and continuing spans/traces from offline (ie log-based) storage. The `Self` reference
bypasses trace and span id generation.
Usage requires passing in a `SpanContext` and the jaeger `Self` reference type:
```
span := tracer.StartSpan(
"continued_span",
SelfRef(yourSpanContext),
)
...
defer span.finish()
```
## License
[Apache 2.0 License](LICENSE).
vendor/github.com/uber/jaeger-client-go/config/config.go
Generated Vendored
+29 -17
View File
@@ -76,16 +76,28 @@ type SamplerConfig struct {
// Can be set by exporting an environment variable named JAEGER_SAMPLER_MANAGER_HOST_PORT
SamplingServerURL string `yaml:"samplingServerURL"`
// MaxOperations is the maximum number of operations that the sampler
// will keep track of. If an operation is not tracked, a default probabilistic
// sampler will be used rather than the per operation specific sampler.
// Can be set by exporting an environment variable named JAEGER_SAMPLER_MAX_OPERATIONS
MaxOperations int `yaml:"maxOperations"`
// SamplingRefreshInterval controls how often the remotely controlled sampler will poll
// jaeger-agent for the appropriate sampling strategy.
// Can be set by exporting an environment variable named JAEGER_SAMPLER_REFRESH_INTERVAL
SamplingRefreshInterval time.Duration `yaml:"samplingRefreshInterval"`
// MaxOperations is the maximum number of operations that the PerOperationSampler
// will keep track of. If an operation is not tracked, a default probabilistic
// sampler will be used rather than the per operation specific sampler.
// Can be set by exporting an environment variable named JAEGER_SAMPLER_MAX_OPERATIONS.
MaxOperations int `yaml:"maxOperations"`
// Opt-in feature for applications that require late binding of span name via explicit
// call to SetOperationName when using PerOperationSampler. When this feature is enabled,
// the sampler will return retryable=true from OnCreateSpan(), thus leaving the sampling
// decision as non-final (and the span as writeable). This may lead to degraded performance
// in applications that always provide the correct span name on trace creation.
//
// For backwards compatibility this option is off by default.
OperationNameLateBinding bool `yaml:"operationNameLateBinding"`
// Options can be used to programmatically pass additional options to the Remote sampler.
Options []jaeger.SamplerOption
}
// ReporterConfig configures the reporter. All fields are optional.
@@ -181,13 +193,14 @@ func (c Configuration) New(
// NewTracer returns a new tracer based on the current configuration, using the given options,
// and a closer func that can be used to flush buffers before shutdown.
func (c Configuration) NewTracer(options ...Option) (opentracing.Tracer, io.Closer, error) {
if c.Disabled {
return &opentracing.NoopTracer{}, &nullCloser{}, nil
}
if c.ServiceName == "" {
return nil, nil, errors.New("no service name provided")
}
if c.Disabled {
return &opentracing.NoopTracer{}, &nullCloser{}, nil
}
opts := applyOptions(options...)
tracerMetrics := jaeger.NewMetrics(opts.metrics, nil)
if c.RPCMetrics {
@@ -234,6 +247,7 @@ func (c Configuration) NewTracer(options ...Option) (opentracing.Tracer, io.Clos
jaeger.TracerOptions.PoolSpans(opts.poolSpans),
jaeger.TracerOptions.ZipkinSharedRPCSpan(opts.zipkinSharedRPCSpan),
jaeger.TracerOptions.MaxTagValueLength(opts.maxTagValueLength),
jaeger.TracerOptions.NoDebugFlagOnForcedSampling(opts.noDebugFlagOnForcedSampling),
}
for _, tag := range opts.tags {
@@ -330,7 +344,7 @@ func (sc *SamplerConfig) NewSampler(
return jaeger.NewProbabilisticSampler(sc.Param)
}
return nil, fmt.Errorf(
"Invalid Param for probabilistic sampler: %v. Expecting value between 0 and 1",
"invalid Param for probabilistic sampler; expecting value between 0 and 1, received %v",
sc.Param,
)
}
@@ -348,16 +362,14 @@ func (sc *SamplerConfig) NewSampler(
jaeger.SamplerOptions.Metrics(metrics),
jaeger.SamplerOptions.InitialSampler(initSampler),
jaeger.SamplerOptions.SamplingServerURL(sc.SamplingServerURL),
jaeger.SamplerOptions.MaxOperations(sc.MaxOperations),
jaeger.SamplerOptions.OperationNameLateBinding(sc.OperationNameLateBinding),
jaeger.SamplerOptions.SamplingRefreshInterval(sc.SamplingRefreshInterval),
}
if sc.MaxOperations != 0 {
options = append(options, jaeger.SamplerOptions.MaxOperations(sc.MaxOperations))
}
if sc.SamplingRefreshInterval != 0 {
options = append(options, jaeger.SamplerOptions.SamplingRefreshInterval(sc.SamplingRefreshInterval))
}
options = append(options, sc.Options...)
return jaeger.NewRemotelyControlledSampler(serviceName, options...), nil
}
return nil, fmt.Errorf("Unknown sampler type %v", sc.Type)
return nil, fmt.Errorf("unknown sampler type (%s)", sc.Type)
}
// NewReporter instantiates a new reporter that submits spans to the collector
vendor/github.com/uber/jaeger-client-go/config/config_env.go
Generated Vendored
+19 -8
View File
@@ -52,7 +52,11 @@ const (
// FromEnv uses environment variables to set the tracer's Configuration
func FromEnv() (*Configuration, error) {
c := &Configuration{}
return c.FromEnv()
}
// FromEnv uses environment variables and overrides existing tracer's Configuration
func (c *Configuration) FromEnv() (*Configuration, error) {
if e := os.Getenv(envServiceName); e != "" {
c.ServiceName = e
}
@@ -77,13 +81,21 @@ func FromEnv() (*Configuration, error) {
c.Tags = parseTags(e)
}
if s, err := samplerConfigFromEnv(); err == nil {
if c.Sampler == nil {
c.Sampler = &SamplerConfig{}
}
if s, err := c.Sampler.samplerConfigFromEnv(); err == nil {
c.Sampler = s
} else {
return nil, errors.Wrap(err, "cannot obtain sampler config from env")
}
if r, err := reporterConfigFromEnv(); err == nil {
if c.Reporter == nil {
c.Reporter = &ReporterConfig{}
}
if r, err := c.Reporter.reporterConfigFromEnv(); err == nil {
c.Reporter = r
} else {
return nil, errors.Wrap(err, "cannot obtain reporter config from env")
@@ -93,9 +105,7 @@ func FromEnv() (*Configuration, error) {
}
// samplerConfigFromEnv creates a new SamplerConfig based on the environment variables
func samplerConfigFromEnv() (*SamplerConfig, error) {
sc := &SamplerConfig{}
func (sc *SamplerConfig) samplerConfigFromEnv() (*SamplerConfig, error) {
if e := os.Getenv(envSamplerType); e != "" {
sc.Type = e
}
@@ -110,6 +120,9 @@ func samplerConfigFromEnv() (*SamplerConfig, error) {
if e := os.Getenv(envSamplerManagerHostPort); e != "" {
sc.SamplingServerURL = e
} else if e := os.Getenv(envAgentHost); e != "" {
// Fallback if we know the agent host - try the sampling endpoint there
sc.SamplingServerURL = fmt.Sprintf("http://%s:%d/sampling", e, jaeger.DefaultSamplingServerPort)
}
if e := os.Getenv(envSamplerMaxOperations); e != "" {
@@ -132,9 +145,7 @@ func samplerConfigFromEnv() (*SamplerConfig, error) {
}
// reporterConfigFromEnv creates a new ReporterConfig based on the environment variables
func reporterConfigFromEnv() (*ReporterConfig, error) {
rc := &ReporterConfig{}
func (rc *ReporterConfig) reporterConfigFromEnv() (*ReporterConfig, error) {
if e := os.Getenv(envReporterMaxQueueSize); e != "" {
if value, err := strconv.ParseInt(e, 10, 0); err == nil {
rc.QueueSize = int(value)
vendor/github.com/uber/jaeger-client-go/config/options.go
Generated Vendored
+22 -13
View File
@@ -26,19 +26,20 @@ type Option func(c *Options)
// Options control behavior of the client.
type Options struct {
metrics metrics.Factory
logger jaeger.Logger
reporter jaeger.Reporter
sampler jaeger.Sampler
contribObservers []jaeger.ContribObserver
observers []jaeger.Observer
gen128Bit bool
poolSpans bool
zipkinSharedRPCSpan bool
maxTagValueLength int
tags []opentracing.Tag
injectors map[interface{}]jaeger.Injector
extractors map[interface{}]jaeger.Extractor
metrics metrics.Factory
logger jaeger.Logger
reporter jaeger.Reporter
sampler jaeger.Sampler
contribObservers []jaeger.ContribObserver
observers []jaeger.Observer
gen128Bit bool
poolSpans bool
zipkinSharedRPCSpan bool
maxTagValueLength int
noDebugFlagOnForcedSampling bool
tags []opentracing.Tag
injectors map[interface{}]jaeger.Injector
extractors map[interface{}]jaeger.Extractor
}
// Metrics creates an Option that initializes Metrics in the tracer,
@@ -117,6 +118,14 @@ func MaxTagValueLength(maxTagValueLength int) Option {
}
}
// NoDebugFlagOnForcedSampling can be used to decide whether debug flag will be set or not
// when calling span.setSamplingPriority to force sample a span.
func NoDebugFlagOnForcedSampling(noDebugFlagOnForcedSampling bool) Option {
return func(c *Options) {
c.noDebugFlagOnForcedSampling = noDebugFlagOnForcedSampling
}
}
// Tag creates an option that adds a tracer-level tag.
func Tag(key string, value interface{}) Option {
return func(c *Options) {
vendor/github.com/uber/jaeger-client-go/constants.go
Generated Vendored
+19 -1
View File
@@ -14,9 +14,15 @@
package jaeger
import (
"fmt"
"github.com/opentracing/opentracing-go"
)
const (
// JaegerClientVersion is the version of the client library reported as Span tag.
JaegerClientVersion = "Go-2.16.0"
JaegerClientVersion = "Go-2.20.1"
// JaegerClientVersionTagKey is the name of the tag used to report client version.
JaegerClientVersionTagKey = "jaeger.version"
@@ -83,6 +89,18 @@ const (
// DefaultUDPSpanServerPort is the default port to send the spans to, via UDP
DefaultUDPSpanServerPort = 6831
// DefaultSamplingServerPort is the default port to fetch sampling config from, via http
DefaultSamplingServerPort = 5778
// DefaultMaxTagValueLength is the default max length of byte array or string allowed in the tag value.
DefaultMaxTagValueLength = 256
// SelfRefType is a jaeger specific reference type that supports creating a span
// with an already defined context.
selfRefType opentracing.SpanReferenceType = 99
)
var (
// DefaultSamplingServerURL is the default url to fetch sampling config from, via http
DefaultSamplingServerURL = fmt.Sprintf("http://localhost:%d/sampling", DefaultSamplingServerPort)
)
vendor/github.com/uber/jaeger-client-go/glide.lock
Generated Vendored
+26 -20
View File
@@ -1,8 +1,8 @@
hash: 92cc8f956428fc65bee07d809a752f34376aece141c934eff02aefa08d450b72
updated: 2019-03-23T18:26:09.960887-04:00
hash: a4a449cfc060c2d7be850a69b171e4382a3bd00d1a0a72cfc944facc3fe263bf
updated: 2019-09-23T17:10:15.213856-04:00
imports:
- name: github.com/beorn7/perks
version: 3a771d992973f24aa725d07868b467d1ddfceafb
version: 37c8de3658fcb183f997c4e13e8337516ab753e6
subpackages:
- quantile
- name: github.com/codahale/hdrhistogram
@@ -17,11 +17,11 @@ imports:
subpackages:
- spew
- name: github.com/golang/protobuf
version: bbd03ef6da3a115852eaf24c8a1c46aeb39aa175
version: 1680a479a2cfb3fa22b972af7e36d0a0fde47bf8
subpackages:
- proto
- name: github.com/matttproud/golang_protobuf_extensions
version: c12348ce28de40eed0136aa2b644d0ee0650e56c
version: c182affec369e30f25d3eb8cd8a478dee585ae7d
subpackages:
- pbutil
- name: github.com/opentracing/opentracing-go
@@ -33,47 +33,49 @@ imports:
- name: github.com/pkg/errors
version: ba968bfe8b2f7e042a574c888954fccecfa385b4
- name: github.com/pmezard/go-difflib
version: 792786c7400a136282c1664665ae0a8db921c6c2
version: 5d4384ee4fb2527b0a1256a821ebfc92f91efefc
subpackages:
- difflib
- name: github.com/prometheus/client_golang
version: c5b7fccd204277076155f10851dad72b76a49317
version: 170205fb58decfd011f1550d4cfb737230d7ae4f
subpackages:
- prometheus
- prometheus/internal
- name: github.com/prometheus/client_model
version: 99fa1f4be8e564e8a6b613da7fa6f46c9edafc6c
version: 14fe0d1b01d4d5fc031dd4bec1823bd3ebbe8016
subpackages:
- go
- name: github.com/prometheus/common
version: 38c53a9f4bfcd932d1b00bfc65e256a7fba6b37a
version: 287d3e634a1e550c9e463dd7e5a75a422c614505
subpackages:
- expfmt
- internal/bitbucket.org/ww/goautoneg
- model
- name: github.com/prometheus/procfs
version: 780932d4fbbe0e69b84c34c20f5c8d0981e109ea
version: de25ac347ef9305868b04dc42425c973b863b18c
subpackages:
- internal/fs
- internal/util
- nfs
- xfs
- name: github.com/stretchr/testify
version: f35b8ab0b5a2cef36673838d662e249dd9c94686
version: 85f2b59c4459e5bf57488796be8c3667cb8246d6
subpackages:
- assert
- require
- suite
- name: github.com/uber-go/atomic
version: df976f2515e274675050de7b3f42545de80594fd
- name: github.com/uber/jaeger-lib
version: 0e30338a695636fe5bcf7301e8030ce8dd2a8530
version: a87ae9d84fb038a8d79266298970720be7c80fcd
subpackages:
- metrics
- metrics/metricstest
- metrics/prometheus
- name: go.uber.org/atomic
version: 1ea20fb1cbb1cc08cbd0d913a96dead89aa18289
version: df976f2515e274675050de7b3f42545de80594fd
- name: go.uber.org/multierr
version: 3c4937480c32f4c13a875a1829af76c98ca3d40a
- name: go.uber.org/zap
version: ff33455a0e382e8a81d14dd7c922020b6b5e7982
version: 27376062155ad36be76b0f12cf1572a221d3a48c
subpackages:
- buffer
- internal/bufferpool
@@ -81,10 +83,14 @@ imports:
- internal/exit
- zapcore
- name: golang.org/x/net
version: 49bb7cea24b1df9410e1712aa6433dae904ff66a
version: aa69164e4478b84860dc6769c710c699c67058a3
subpackages:
- context
- context/ctxhttp
testImports:
- name: github.com/uber-go/atomic
version: 8474b86a5a6f79c443ce4b2992817ff32cf208b8
- name: golang.org/x/sys
version: 0a153f010e6963173baba2306531d173aa843137
subpackages:
- windows
- name: gopkg.in/yaml.v2
version: 51d6538a90f86fe93ac480b35f37b2be17fef232
testImports: []
vendor/github.com/uber/jaeger-client-go/glide.yaml
Generated Vendored
+7 -2
View File
@@ -12,11 +12,16 @@ import:
- metrics
- package: github.com/pkg/errors
version: ~0.8.0
- package: go.uber.org/zap
source: https://github.com/uber-go/zap.git
version: ^1
- package: github.com/uber-go/atomic
version: ^1
- package: github.com/prometheus/client_golang
version: ^1
testImport:
- package: github.com/stretchr/testify
subpackages:
- assert
- require
- suite
- package: github.com/prometheus/client_golang
version: v0.8.0
vendor/github.com/uber/jaeger-client-go/jaeger_thrift_span.go
Generated Vendored
+1 -1
View File
@@ -35,7 +35,7 @@ func BuildJaegerThrift(span *Span) *j.Span {
SpanId: int64(span.context.spanID),
ParentSpanId: int64(span.context.parentID),
OperationName: span.operationName,
Flags: int32(span.context.flags),
Flags: int32(span.context.samplingState.flags()),
StartTime: startTime,
Duration: duration,
Tags: buildTags(span.tags, span.tracer.options.maxTagValueLength),
vendor/github.com/uber/jaeger-client-go/metrics.go
Generated Vendored
+16 -4
View File
@@ -26,6 +26,9 @@ type Metrics struct {
// Number of traces started by this tracer as not sampled
TracesStartedNotSampled metrics.Counter `metric:"traces" tags:"state=started,sampled=n" help:"Number of traces started by this tracer as not sampled"`
// Number of traces started by this tracer with delayed sampling
TracesStartedDelayedSampling metrics.Counter `metric:"traces" tags:"state=started,sampled=n" help:"Number of traces started by this tracer with delayed sampling"`
// Number of externally started sampled traces this tracer joined
TracesJoinedSampled metrics.Counter `metric:"traces" tags:"state=joined,sampled=y" help:"Number of externally started sampled traces this tracer joined"`
@@ -33,13 +36,22 @@ type Metrics struct {
TracesJoinedNotSampled metrics.Counter `metric:"traces" tags:"state=joined,sampled=n" help:"Number of externally started not-sampled traces this tracer joined"`
// Number of sampled spans started by this tracer
SpansStartedSampled metrics.Counter `metric:"started_spans" tags:"sampled=y" help:"Number of sampled spans started by this tracer"`
SpansStartedSampled metrics.Counter `metric:"started_spans" tags:"sampled=y" help:"Number of spans started by this tracer as sampled"`
// Number of unsampled spans started by this tracer
SpansStartedNotSampled metrics.Counter `metric:"started_spans" tags:"sampled=n" help:"Number of unsampled spans started by this tracer"`
// Number of not sampled spans started by this tracer
SpansStartedNotSampled metrics.Counter `metric:"started_spans" tags:"sampled=n" help:"Number of spans started by this tracer as not sampled"`
// Number of spans with delayed sampling started by this tracer
SpansStartedDelayedSampling metrics.Counter `metric:"started_spans" tags:"sampled=delayed" help:"Number of spans started by this tracer with delayed sampling"`
// Number of spans finished by this tracer
SpansFinished metrics.Counter `metric:"finished_spans" help:"Number of spans finished by this tracer"`
SpansFinishedSampled metrics.Counter `metric:"finished_spans" tags:"sampled=y" help:"Number of sampled spans finished by this tracer"`
// Number of spans finished by this tracer
SpansFinishedNotSampled metrics.Counter `metric:"finished_spans" tags:"sampled=n" help:"Number of not-sampled spans finished by this tracer"`
// Number of spans finished by this tracer
SpansFinishedDelayedSampling metrics.Counter `metric:"finished_spans" tags:"sampled=delayed" help:"Number of spans with delayed sampling finished by this tracer"`
// Number of errors decoding tracing context
DecodingErrors metrics.Counter `metric:"span_context_decoding_errors" help:"Number of errors decoding tracing context"`
vendor/github.com/uber/jaeger-client-go/propagation.go
Generated Vendored
+6 -2
View File
@@ -193,7 +193,7 @@ func (p *BinaryPropagator) Inject(
if err := binary.Write(carrier, binary.BigEndian, sc.parentID); err != nil {
return err
}
if err := binary.Write(carrier, binary.BigEndian, sc.flags); err != nil {
if err := binary.Write(carrier, binary.BigEndian, sc.samplingState.flags()); err != nil {
return err
}
@@ -222,6 +222,7 @@ func (p *BinaryPropagator) Extract(abstractCarrier interface{}) (SpanContext, er
return emptyContext, opentracing.ErrInvalidCarrier
}
var ctx SpanContext
ctx.samplingState = &samplingState{}
if err := binary.Read(carrier, binary.BigEndian, &ctx.traceID); err != nil {
return emptyContext, opentracing.ErrSpanContextCorrupted
@@ -232,9 +233,12 @@ func (p *BinaryPropagator) Extract(abstractCarrier interface{}) (SpanContext, er
if err := binary.Read(carrier, binary.BigEndian, &ctx.parentID); err != nil {
return emptyContext, opentracing.ErrSpanContextCorrupted
}
if err := binary.Read(carrier, binary.BigEndian, &ctx.flags); err != nil {
var flags byte
if err := binary.Read(carrier, binary.BigEndian, &flags); err != nil {
return emptyContext, opentracing.ErrSpanContextCorrupted
}
ctx.samplingState.setFlags(flags)
// Handle the baggage items
var numBaggage int32
vendor/github.com/uber/jaeger-client-go/reporter.go
Generated Vendored
+15 -5
View File
@@ -28,6 +28,8 @@ import (
// Reporter is called by the tracer when a span is completed to report the span to the tracing collector.
type Reporter interface {
// Report submits a new span to collectors, possibly asynchronously and/or with buffering.
// If the reporter is processing Span asynchronously then it needs to Retain() the span,
// and then Release() it when no longer needed, to avoid span data corruption.
Report(span *Span)
// Close does a clean shutdown of the reporter, flushing any traces that may be buffered in memory.
@@ -93,13 +95,14 @@ func NewInMemoryReporter() *InMemoryReporter {
// Report implements Report() method of Reporter by storing the span in the buffer.
func (r *InMemoryReporter) Report(span *Span) {
r.lock.Lock()
r.spans = append(r.spans, span)
// Need to retain the span otherwise it will be released
r.spans = append(r.spans, span.Retain())
r.lock.Unlock()
}
// Close implements Close() method of Reporter by doing nothing.
// Close implements Close() method of Reporter
func (r *InMemoryReporter) Close() {
// no-op
r.Reset()
}
// SpansSubmitted returns the number of spans accumulated in the buffer.
@@ -122,7 +125,12 @@ func (r *InMemoryReporter) GetSpans() []opentracing.Span {
func (r *InMemoryReporter) Reset() {
r.lock.Lock()
defer r.lock.Unlock()
r.spans = nil
// Before reset the collection need to release Span memory
for _, span := range r.spans {
span.(*Span).Release()
}
r.spans = r.spans[:0]
}
// ------------------------------
@@ -218,7 +226,8 @@ func NewRemoteReporter(sender Transport, opts ...ReporterOption) Reporter {
// because some of them may still be successfully added to the queue.
func (r *remoteReporter) Report(span *Span) {
select {
case r.queue <- reporterQueueItem{itemType: reporterQueueItemSpan, span: span}:
// Need to retain the span otherwise it will be released
case r.queue <- reporterQueueItem{itemType: reporterQueueItemSpan, span: span.Retain()}:
atomic.AddInt64(&r.queueLength, 1)
default:
r.metrics.ReporterDropped.Inc(1)
@@ -278,6 +287,7 @@ func (r *remoteReporter) processQueue() {
// to reduce the number of gauge stats, we only emit queue length on flush
r.metrics.ReporterQueueLength.Update(atomic.LoadInt64(&r.queueLength))
}
span.Release()
case reporterQueueItemClose:
timer.Stop()
flush()
vendor/github.com/uber/jaeger-client-go/sampler.go
Generated Vendored
+196 -259
View File
@@ -17,20 +17,14 @@ package jaeger
import (
"fmt"
"math"
"net/url"
"sync"
"sync/atomic"
"time"
"github.com/uber/jaeger-client-go/log"
"github.com/uber/jaeger-client-go/thrift-gen/sampling"
"github.com/uber/jaeger-client-go/utils"
)
const (
defaultSamplingServerURL = "http://localhost:5778/sampling"
defaultSamplingRefreshInterval = time.Minute
defaultMaxOperations = 2000
defaultMaxOperations = 2000
)
// Sampler decides whether a new trace should be sampled or not.
@@ -48,9 +42,7 @@ type Sampler interface {
// Equal checks if the `other` sampler is functionally equivalent
// to this sampler.
// TODO remove this function. This function is used to determine if 2 samplers are equivalent
// which does not bode well with the adaptive sampler which has to create all the composite samplers
// for the comparison to occur. This is expensive to do if only one sampler has changed.
// TODO (breaking change) remove this function. See PerOperationSampler.Equals for explanation.
Equal(other Sampler) bool
}
@@ -58,17 +50,23 @@ type Sampler interface {
// ConstSampler is a sampler that always makes the same decision.
type ConstSampler struct {
legacySamplerV1Base
Decision bool
tags []Tag
}
// NewConstSampler creates a ConstSampler.
func NewConstSampler(sample bool) Sampler {
func NewConstSampler(sample bool) *ConstSampler {
tags := []Tag{
{key: SamplerTypeTagKey, value: SamplerTypeConst},
{key: SamplerParamTagKey, value: sample},
}
return &ConstSampler{Decision: sample, tags: tags}
s := &ConstSampler{
Decision: sample,
tags: tags,
}
s.delegate = s.IsSampled
return s
}
// IsSampled implements IsSampled() of Sampler.
@@ -89,11 +87,17 @@ func (s *ConstSampler) Equal(other Sampler) bool {
return false
}
// String is used to log sampler details.
func (s *ConstSampler) String() string {
return fmt.Sprintf("ConstSampler(decision=%t)", s.Decision)
}
// -----------------------
// ProbabilisticSampler is a sampler that randomly samples a certain percentage
// of traces.
type ProbabilisticSampler struct {
legacySamplerV1Base
samplingRate float64
samplingBoundary uint64
tags []Tag
@@ -115,16 +119,19 @@ func NewProbabilisticSampler(samplingRate float64) (*ProbabilisticSampler, error
}
func newProbabilisticSampler(samplingRate float64) *ProbabilisticSampler {
samplingRate = math.Max(0.0, math.Min(samplingRate, 1.0))
tags := []Tag{
s := new(ProbabilisticSampler)
s.delegate = s.IsSampled
return s.init(samplingRate)
}
func (s *ProbabilisticSampler) init(samplingRate float64) *ProbabilisticSampler {
s.samplingRate = math.Max(0.0, math.Min(samplingRate, 1.0))
s.samplingBoundary = uint64(float64(maxRandomNumber) * s.samplingRate)
s.tags = []Tag{
{key: SamplerTypeTagKey, value: SamplerTypeProbabilistic},
{key: SamplerParamTagKey, value: samplingRate},
}
return &ProbabilisticSampler{
samplingRate: samplingRate,
samplingBoundary: uint64(float64(maxRandomNumber) * samplingRate),
tags: tags,
{key: SamplerParamTagKey, value: s.samplingRate},
}
return s
}
// SamplingRate returns the sampling probability this sampled was constructed with.
@@ -150,65 +157,104 @@ func (s *ProbabilisticSampler) Equal(other Sampler) bool {
return false
}
// Update modifies in-place the sampling rate. Locking must be done externally.
func (s *ProbabilisticSampler) Update(samplingRate float64) error {
if samplingRate < 0.0 || samplingRate > 1.0 {
return fmt.Errorf("Sampling Rate must be between 0.0 and 1.0, received %f", samplingRate)
}
s.init(samplingRate)
return nil
}
// String is used to log sampler details.
func (s *ProbabilisticSampler) String() string {
return fmt.Sprintf("ProbabilisticSampler(samplingRate=%v)", s.samplingRate)
}
// -----------------------
type rateLimitingSampler struct {
// RateLimitingSampler samples at most maxTracesPerSecond. The distribution of sampled traces follows
// burstiness of the service, i.e. a service with uniformly distributed requests will have those
// requests sampled uniformly as well, but if requests are bursty, especially sub-second, then a
// number of sequential requests can be sampled each second.
type RateLimitingSampler struct {
legacySamplerV1Base
maxTracesPerSecond float64
rateLimiter utils.RateLimiter
rateLimiter *utils.ReconfigurableRateLimiter
tags []Tag
}
// NewRateLimitingSampler creates a sampler that samples at most maxTracesPerSecond. The distribution of sampled
// traces follows burstiness of the service, i.e. a service with uniformly distributed requests will have those
// requests sampled uniformly as well, but if requests are bursty, especially sub-second, then a number of
// sequential requests can be sampled each second.
func NewRateLimitingSampler(maxTracesPerSecond float64) Sampler {
tags := []Tag{
// NewRateLimitingSampler creates new RateLimitingSampler.
func NewRateLimitingSampler(maxTracesPerSecond float64) *RateLimitingSampler {
s := new(RateLimitingSampler)
s.delegate = s.IsSampled
return s.init(maxTracesPerSecond)
}
func (s *RateLimitingSampler) init(maxTracesPerSecond float64) *RateLimitingSampler {
if s.rateLimiter == nil {
s.rateLimiter = utils.NewRateLimiter(maxTracesPerSecond, math.Max(maxTracesPerSecond, 1.0))
} else {
s.rateLimiter.Update(maxTracesPerSecond, math.Max(maxTracesPerSecond, 1.0))
}
s.maxTracesPerSecond = maxTracesPerSecond
s.tags = []Tag{
{key: SamplerTypeTagKey, value: SamplerTypeRateLimiting},
{key: SamplerParamTagKey, value: maxTracesPerSecond},
}
return &rateLimitingSampler{
maxTracesPerSecond: maxTracesPerSecond,
rateLimiter: utils.NewRateLimiter(maxTracesPerSecond, math.Max(maxTracesPerSecond, 1.0)),
tags: tags,
}
return s
}
// IsSampled implements IsSampled() of Sampler.
func (s *rateLimitingSampler) IsSampled(id TraceID, operation string) (bool, []Tag) {
func (s *RateLimitingSampler) IsSampled(id TraceID, operation string) (bool, []Tag) {
return s.rateLimiter.CheckCredit(1.0), s.tags
}
func (s *rateLimitingSampler) Close() {
// Update reconfigures the rate limiter, while preserving its accumulated balance.
// Locking must be done externally.
func (s *RateLimitingSampler) Update(maxTracesPerSecond float64) {
if s.maxTracesPerSecond != maxTracesPerSecond {
s.init(maxTracesPerSecond)
}
}
// Close does nothing.
func (s *RateLimitingSampler) Close() {
// nothing to do
}
func (s *rateLimitingSampler) Equal(other Sampler) bool {
if o, ok := other.(*rateLimitingSampler); ok {
// Equal compares with another sampler.
func (s *RateLimitingSampler) Equal(other Sampler) bool {
if o, ok := other.(*RateLimitingSampler); ok {
return s.maxTracesPerSecond == o.maxTracesPerSecond
}
return false
}
// String is used to log sampler details.
func (s *RateLimitingSampler) String() string {
return fmt.Sprintf("RateLimitingSampler(maxTracesPerSecond=%v)", s.maxTracesPerSecond)
}
// -----------------------
// GuaranteedThroughputProbabilisticSampler is a sampler that leverages both probabilisticSampler and
// rateLimitingSampler. The rateLimitingSampler is used as a guaranteed lower bound sampler such that
// GuaranteedThroughputProbabilisticSampler is a sampler that leverages both ProbabilisticSampler and
// RateLimitingSampler. The RateLimitingSampler is used as a guaranteed lower bound sampler such that
// every operation is sampled at least once in a time interval defined by the lowerBound. ie a lowerBound
// of 1.0 / (60 * 10) will sample an operation at least once every 10 minutes.
//
// The probabilisticSampler is given higher priority when tags are emitted, ie. if IsSampled() for both
// samplers return true, the tags for probabilisticSampler will be used.
// The ProbabilisticSampler is given higher priority when tags are emitted, ie. if IsSampled() for both
// samplers return true, the tags for ProbabilisticSampler will be used.
type GuaranteedThroughputProbabilisticSampler struct {
probabilisticSampler *ProbabilisticSampler
lowerBoundSampler Sampler
lowerBoundSampler *RateLimitingSampler
tags []Tag
samplingRate float64
lowerBound float64
}
// NewGuaranteedThroughputProbabilisticSampler returns a delegating sampler that applies both
// probabilisticSampler and rateLimitingSampler.
// ProbabilisticSampler and RateLimitingSampler.
func NewGuaranteedThroughputProbabilisticSampler(
lowerBound, samplingRate float64,
) (*GuaranteedThroughputProbabilisticSampler, error) {
@@ -225,8 +271,14 @@ func newGuaranteedThroughputProbabilisticSampler(lowerBound, samplingRate float6
}
func (s *GuaranteedThroughputProbabilisticSampler) setProbabilisticSampler(samplingRate float64) {
if s.probabilisticSampler == nil || s.samplingRate != samplingRate {
if s.probabilisticSampler == nil {
s.probabilisticSampler = newProbabilisticSampler(samplingRate)
} else if s.samplingRate != samplingRate {
s.probabilisticSampler.init(samplingRate)
}
// since we don't validate samplingRate, sampler may have clamped it to [0, 1] interval
samplingRate = s.probabilisticSampler.SamplingRate()
if s.samplingRate != samplingRate || s.tags == nil {
s.samplingRate = s.probabilisticSampler.SamplingRate()
s.tags = []Tag{
{key: SamplerTypeTagKey, value: SamplerTypeLowerBound},
@@ -253,7 +305,7 @@ func (s *GuaranteedThroughputProbabilisticSampler) Close() {
// Equal implements Equal() of Sampler.
func (s *GuaranteedThroughputProbabilisticSampler) Equal(other Sampler) bool {
// NB The Equal() function is expensive and will be removed. See adaptiveSampler.Equal() for
// NB The Equal() function is expensive and will be removed. See PerOperationSampler.Equal() for
// more information.
return false
}
@@ -262,52 +314,119 @@ func (s *GuaranteedThroughputProbabilisticSampler) Equal(other Sampler) bool {
func (s *GuaranteedThroughputProbabilisticSampler) update(lowerBound, samplingRate float64) {
s.setProbabilisticSampler(samplingRate)
if s.lowerBound != lowerBound {
s.lowerBoundSampler = NewRateLimitingSampler(lowerBound)
s.lowerBoundSampler.Update(lowerBound)
s.lowerBound = lowerBound
}
}
// -----------------------
type adaptiveSampler struct {
// PerOperationSampler is a delegating sampler that applies GuaranteedThroughputProbabilisticSampler
// on a per-operation basis.
type PerOperationSampler struct {
sync.RWMutex
samplers map[string]*GuaranteedThroughputProbabilisticSampler
defaultSampler *ProbabilisticSampler
lowerBound float64
maxOperations int
// see description in PerOperationSamplerParams
operationNameLateBinding bool
}
// NewAdaptiveSampler returns a delegating sampler that applies both probabilisticSampler and
// rateLimitingSampler via the guaranteedThroughputProbabilisticSampler. This sampler keeps track of all
// operations and delegates calls to the respective guaranteedThroughputProbabilisticSampler.
func NewAdaptiveSampler(strategies *sampling.PerOperationSamplingStrategies, maxOperations int) (Sampler, error) {
return newAdaptiveSampler(strategies, maxOperations), nil
// NewAdaptiveSampler returns a new PerOperationSampler.
// Deprecated: please use NewPerOperationSampler.
func NewAdaptiveSampler(strategies *sampling.PerOperationSamplingStrategies, maxOperations int) (*PerOperationSampler, error) {
return NewPerOperationSampler(PerOperationSamplerParams{
MaxOperations: maxOperations,
Strategies: strategies,
}), nil
}
func newAdaptiveSampler(strategies *sampling.PerOperationSamplingStrategies, maxOperations int) Sampler {
// PerOperationSamplerParams defines parameters when creating PerOperationSampler.
type PerOperationSamplerParams struct {
// Max number of operations that will be tracked. Other operations will be given default strategy.
MaxOperations int
// Opt-in feature for applications that require late binding of span name via explicit call to SetOperationName.
// When this feature is enabled, the sampler will return retryable=true from OnCreateSpan(), thus leaving
// the sampling decision as non-final (and the span as writeable). This may lead to degraded performance
// in applications that always provide the correct span name on trace creation.
//
// For backwards compatibility this option is off by default.
OperationNameLateBinding bool
// Initial configuration of the sampling strategies (usually retrieved from the backend by Remote Sampler).
Strategies *sampling.PerOperationSamplingStrategies
}
// NewPerOperationSampler returns a new PerOperationSampler.
func NewPerOperationSampler(params PerOperationSamplerParams) *PerOperationSampler {
if params.MaxOperations <= 0 {
params.MaxOperations = defaultMaxOperations
}
samplers := make(map[string]*GuaranteedThroughputProbabilisticSampler)
for _, strategy := range strategies.PerOperationStrategies {
for _, strategy := range params.Strategies.PerOperationStrategies {
sampler := newGuaranteedThroughputProbabilisticSampler(
strategies.DefaultLowerBoundTracesPerSecond,
params.Strategies.DefaultLowerBoundTracesPerSecond,
strategy.ProbabilisticSampling.SamplingRate,
)
samplers[strategy.Operation] = sampler
}
return &adaptiveSampler{
samplers: samplers,
defaultSampler: newProbabilisticSampler(strategies.DefaultSamplingProbability),
lowerBound: strategies.DefaultLowerBoundTracesPerSecond,
maxOperations: maxOperations,
return &PerOperationSampler{
samplers: samplers,
defaultSampler: newProbabilisticSampler(params.Strategies.DefaultSamplingProbability),
lowerBound: params.Strategies.DefaultLowerBoundTracesPerSecond,
maxOperations: params.MaxOperations,
operationNameLateBinding: params.OperationNameLateBinding,
}
}
func (s *adaptiveSampler) IsSampled(id TraceID, operation string) (bool, []Tag) {
// IsSampled is not used and only exists to match Sampler V1 API.
// TODO (breaking change) remove when upgrading everything to SamplerV2
func (s *PerOperationSampler) IsSampled(id TraceID, operation string) (bool, []Tag) {
return false, nil
}
func (s *PerOperationSampler) trySampling(span *Span, operationName string) (bool, []Tag) {
samplerV1 := s.getSamplerForOperation(operationName)
var sampled bool
var tags []Tag
if span.context.samplingState.isLocalRootSpan(span.context.spanID) {
sampled, tags = samplerV1.IsSampled(span.context.TraceID(), operationName)
}
return sampled, tags
}
// OnCreateSpan implements OnCreateSpan of SamplerV2.
func (s *PerOperationSampler) OnCreateSpan(span *Span) SamplingDecision {
sampled, tags := s.trySampling(span, span.OperationName())
return SamplingDecision{Sample: sampled, Retryable: s.operationNameLateBinding, Tags: tags}
}
// OnSetOperationName implements OnSetOperationName of SamplerV2.
func (s *PerOperationSampler) OnSetOperationName(span *Span, operationName string) SamplingDecision {
sampled, tags := s.trySampling(span, operationName)
return SamplingDecision{Sample: sampled, Retryable: false, Tags: tags}
}
// OnSetTag implements OnSetTag of SamplerV2.
func (s *PerOperationSampler) OnSetTag(span *Span, key string, value interface{}) SamplingDecision {
return SamplingDecision{Sample: false, Retryable: true}
}
// OnFinishSpan implements OnFinishSpan of SamplerV2.
func (s *PerOperationSampler) OnFinishSpan(span *Span) SamplingDecision {
return SamplingDecision{Sample: false, Retryable: true}
}
func (s *PerOperationSampler) getSamplerForOperation(operation string) Sampler {
s.RLock()
sampler, ok := s.samplers[operation]
if ok {
defer s.RUnlock()
return sampler.IsSampled(id, operation)
return sampler
}
s.RUnlock()
s.Lock()
@@ -316,18 +435,19 @@ func (s *adaptiveSampler) IsSampled(id TraceID, operation string) (bool, []Tag)
// Check if sampler has already been created
sampler, ok = s.samplers[operation]
if ok {
return sampler.IsSampled(id, operation)
return sampler
}
// Store only up to maxOperations of unique ops.
if len(s.samplers) >= s.maxOperations {
return s.defaultSampler.IsSampled(id, operation)
return s.defaultSampler
}
newSampler := newGuaranteedThroughputProbabilisticSampler(s.lowerBound, s.defaultSampler.SamplingRate())
s.samplers[operation] = newSampler
return newSampler.IsSampled(id, operation)
return newSampler
}
func (s *adaptiveSampler) Close() {
// Close invokes Close on all underlying samplers.
func (s *PerOperationSampler) Close() {
s.Lock()
defer s.Unlock()
for _, sampler := range s.samplers {
@@ -336,222 +456,39 @@ func (s *adaptiveSampler) Close() {
s.defaultSampler.Close()
}
func (s *adaptiveSampler) Equal(other Sampler) bool {
// NB The Equal() function is overly expensive for adaptiveSampler since it's composed of multiple
// Equal is not used.
// TODO (breaking change) remove this in the future
func (s *PerOperationSampler) Equal(other Sampler) bool {
// NB The Equal() function is overly expensive for PerOperationSampler since it's composed of multiple
// samplers which all need to be initialized before this function can be called for a comparison.
// Therefore, adaptiveSampler uses the update() function to only alter the samplers that need
// Therefore, PerOperationSampler uses the update() function to only alter the samplers that need
// changing. Hence this function always returns false so that the update function can be called.
// Once the Equal() function is removed from the Sampler API, this will no longer be needed.
return false
}
func (s *adaptiveSampler) update(strategies *sampling.PerOperationSamplingStrategies) {
func (s *PerOperationSampler) update(strategies *sampling.PerOperationSamplingStrategies) {
s.Lock()
defer s.Unlock()
newSamplers := map[string]*GuaranteedThroughputProbabilisticSampler{}
for _, strategy := range strategies.PerOperationStrategies {
operation := strategy.Operation
samplingRate := strategy.ProbabilisticSampling.SamplingRate
lowerBound := strategies.DefaultLowerBoundTracesPerSecond
if sampler, ok := s.samplers[operation]; ok {
sampler.update(lowerBound, samplingRate)
newSamplers[operation] = sampler
} else {
sampler := newGuaranteedThroughputProbabilisticSampler(
lowerBound,
samplingRate,
)
s.samplers[operation] = sampler
newSamplers[operation] = sampler
}
}
s.lowerBound = strategies.DefaultLowerBoundTracesPerSecond
if s.defaultSampler.SamplingRate() != strategies.DefaultSamplingProbability {
s.defaultSampler = newProbabilisticSampler(strategies.DefaultSamplingProbability)
}
}
// -----------------------
// 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
samplerOptions
serviceName string
manager sampling.SamplingManager
doneChan chan *sync.WaitGroup
}
type httpSamplingManager struct {
serverURL string
}
func (s *httpSamplingManager) GetSamplingStrategy(serviceName string) (*sampling.SamplingStrategyResponse, error) {
var out sampling.SamplingStrategyResponse
v := url.Values{}
v.Set("service", serviceName)
if err := utils.GetJSON(s.serverURL+"?"+v.Encode(), &out); err != nil {
return nil, err
}
return &out, nil
}
// 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 := applySamplerOptions(opts...)
sampler := &RemotelyControlledSampler{
samplerOptions: options,
serviceName: serviceName,
manager: &httpSamplingManager{serverURL: options.samplingServerURL},
doneChan: make(chan *sync.WaitGroup),
}
go sampler.pollController()
return sampler
}
func applySamplerOptions(opts ...SamplerOption) samplerOptions {
options := samplerOptions{}
for _, option := range opts {
option(&options)
}
if options.sampler == nil {
options.sampler = newProbabilisticSampler(0.001)
}
if options.logger == nil {
options.logger = log.NullLogger
}
if options.maxOperations <= 0 {
options.maxOperations = defaultMaxOperations
}
if options.samplingServerURL == "" {
options.samplingServerURL = defaultSamplingServerURL
}
if options.metrics == nil {
options.metrics = NewNullMetrics()
}
if options.samplingRefreshInterval <= 0 {
options.samplingRefreshInterval = defaultSamplingRefreshInterval
}
return options
}
// IsSampled implements IsSampled() of Sampler.
func (s *RemotelyControlledSampler) IsSampled(id TraceID, operation string) (bool, []Tag) {
s.RLock()
defer s.RUnlock()
return s.sampler.IsSampled(id, operation)
}
// 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 adaptiveSampler.Equal() for
// more information.
if o, ok := other.(*RemotelyControlledSampler); ok {
s.RLock()
o.RLock()
defer s.RUnlock()
defer o.RUnlock()
return s.sampler.Equal(o.sampler)
}
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
}
}
}
func (s *RemotelyControlledSampler) getSampler() Sampler {
s.Lock()
defer s.Unlock()
return s.sampler
}
func (s *RemotelyControlledSampler) setSampler(sampler Sampler) {
s.Lock()
defer s.Unlock()
s.sampler = sampler
}
func (s *RemotelyControlledSampler) updateSampler() {
res, err := s.manager.GetSamplingStrategy(s.serviceName)
if err != nil {
s.metrics.SamplerQueryFailure.Inc(1)
s.logger.Infof("Unable to query sampling strategy: %v", err)
return
}
s.Lock()
defer s.Unlock()
s.metrics.SamplerRetrieved.Inc(1)
if strategies := res.GetOperationSampling(); strategies != nil {
s.updateAdaptiveSampler(strategies)
} else {
err = s.updateRateLimitingOrProbabilisticSampler(res)
}
if err != nil {
s.metrics.SamplerUpdateFailure.Inc(1)
s.logger.Infof("Unable 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) updateAdaptiveSampler(strategies *sampling.PerOperationSamplingStrategies) {
if adaptiveSampler, ok := s.sampler.(*adaptiveSampler); ok {
adaptiveSampler.update(strategies)
} else {
s.sampler = newAdaptiveSampler(strategies, s.maxOperations)
}
}
// NB: this function should only be called while holding a Write lock
func (s *RemotelyControlledSampler) updateRateLimitingOrProbabilisticSampler(res *sampling.SamplingStrategyResponse) error {
var newSampler Sampler
if probabilistic := res.GetProbabilisticSampling(); probabilistic != nil {
newSampler = newProbabilisticSampler(probabilistic.SamplingRate)
} else if rateLimiting := res.GetRateLimitingSampling(); rateLimiting != nil {
newSampler = NewRateLimitingSampler(float64(rateLimiting.MaxTracesPerSecond))
} else {
return fmt.Errorf("Unsupported sampling strategy type %v", res.GetStrategyType())
}
if !s.sampler.Equal(newSampler) {
s.sampler = newSampler
}
return nil
s.samplers = newSamplers
}
vendor/github.com/uber/jaeger-client-go/sampler_options.go
Generated Vendored
-81
View File
@@ -1,81 +0,0 @@
// 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 (
"time"
)
// SamplerOption is a function that sets some option on the sampler
type SamplerOption func(options *samplerOptions)
// SamplerOptions is a factory for all available SamplerOption's
var SamplerOptions samplerOptions
type samplerOptions struct {
metrics *Metrics
maxOperations int
sampler Sampler
logger Logger
samplingServerURL string
samplingRefreshInterval time.Duration
}
// Metrics creates a SamplerOption that initializes Metrics on the sampler,
// which is used to emit statistics.
func (samplerOptions) Metrics(m *Metrics) SamplerOption {
return func(o *samplerOptions) {
o.metrics = m
}
}
// MaxOperations creates a SamplerOption that sets the maximum number of
// operations the sampler will keep track of.
func (samplerOptions) MaxOperations(maxOperations int) SamplerOption {
return func(o *samplerOptions) {
o.maxOperations = maxOperations
}
}
// InitialSampler creates a SamplerOption that sets the initial sampler
// to use before a remote sampler is created and used.
func (samplerOptions) InitialSampler(sampler Sampler) SamplerOption {
return func(o *samplerOptions) {
o.sampler = sampler
}
}
// Logger creates a SamplerOption that sets the logger used by the sampler.
func (samplerOptions) Logger(logger Logger) SamplerOption {
return func(o *samplerOptions) {
o.logger = logger
}
}
// SamplingServerURL creates a SamplerOption that sets the sampling server url
// of the local agent that contains the sampling strategies.
func (samplerOptions) SamplingServerURL(samplingServerURL string) SamplerOption {
return func(o *samplerOptions) {
o.samplingServerURL = samplingServerURL
}
}
// SamplingRefreshInterval creates a SamplerOption that sets how often the
// sampler will poll local agent for the appropriate sampling strategy.
func (samplerOptions) SamplingRefreshInterval(samplingRefreshInterval time.Duration) SamplerOption {
return func(o *samplerOptions) {
o.samplingRefreshInterval = samplingRefreshInterval
}
}
vendor/github.com/uber/jaeger-client-go/sampler_remote.go
Generated Vendored
+335
View File
@@ -0,0 +1,335 @@
// 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/thrift-gen/sampling"
)
const (
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
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 {
return s.sampler.OnCreateSpan(span)
}
// OnSetOperationName implements OnSetOperationName of SamplerV2.
func (s *RemotelyControlledSampler) OnSetOperationName(span *Span, operationName string) SamplingDecision {
return s.sampler.OnSetOperationName(span, operationName)
}
// OnSetTag implements OnSetTag of SamplerV2.
func (s *RemotelyControlledSampler) OnSetTag(span *Span, key string, value interface{}) SamplingDecision {
return s.sampler.OnSetTag(span, key, value)
}
// OnFinishSpan implements OnFinishSpan of SamplerV2.
func (s *RemotelyControlledSampler) OnFinishSpan(span *Span) SamplingDecision {
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.Lock()
defer s.Unlock()
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.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 Logger
}
func (f *httpSamplingStrategyFetcher) Fetch(serviceName string) ([]byte, error) {
v := url.Values{}
v.Set("service", serviceName)
uri := f.serverURL + "?" + v.Encode()
// TODO create and reuse http.Client with proper timeout settings, etc.
resp, err := http.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
}
vendor/github.com/uber/jaeger-client-go/sampler_remote_options.go
Generated Vendored
+162
View File
@@ -0,0 +1,162 @@
// 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 (
"time"
"github.com/uber/jaeger-client-go/log"
)
// SamplerOption is a function that sets some option on the sampler
type SamplerOption func(options *samplerOptions)
// SamplerOptions is a factory for all available SamplerOption's.
var SamplerOptions SamplerOptionsFactory
// SamplerOptionsFactory is a factory for all available SamplerOption's.
// The type acts as a namespace for factory functions. It is public to
// make the functions discoverable via godoc. Recommended to be used
// via global SamplerOptions variable.
type SamplerOptionsFactory struct{}
type samplerOptions struct {
metrics *Metrics
sampler SamplerV2
logger Logger
samplingServerURL string
samplingRefreshInterval time.Duration
samplingFetcher SamplingStrategyFetcher
samplingParser SamplingStrategyParser
updaters []SamplerUpdater
posParams PerOperationSamplerParams
}
// Metrics creates a SamplerOption that initializes Metrics on the sampler,
// which is used to emit statistics.
func (SamplerOptionsFactory) Metrics(m *Metrics) SamplerOption {
return func(o *samplerOptions) {
o.metrics = m
}
}
// MaxOperations creates a SamplerOption that sets the maximum number of
// operations the sampler will keep track of.
func (SamplerOptionsFactory) MaxOperations(maxOperations int) SamplerOption {
return func(o *samplerOptions) {
o.posParams.MaxOperations = maxOperations
}
}
// OperationNameLateBinding creates a SamplerOption that sets the respective
// field in the PerOperationSamplerParams.
func (SamplerOptionsFactory) OperationNameLateBinding(enable bool) SamplerOption {
return func(o *samplerOptions) {
o.posParams.OperationNameLateBinding = enable
}
}
// InitialSampler creates a SamplerOption that sets the initial sampler
// to use before a remote sampler is created and used.
func (SamplerOptionsFactory) InitialSampler(sampler Sampler) SamplerOption {
return func(o *samplerOptions) {
o.sampler = samplerV1toV2(sampler)
}
}
// Logger creates a SamplerOption that sets the logger used by the sampler.
func (SamplerOptionsFactory) Logger(logger Logger) SamplerOption {
return func(o *samplerOptions) {
o.logger = logger
}
}
// SamplingServerURL creates a SamplerOption that sets the sampling server url
// of the local agent that contains the sampling strategies.
func (SamplerOptionsFactory) SamplingServerURL(samplingServerURL string) SamplerOption {
return func(o *samplerOptions) {
o.samplingServerURL = samplingServerURL
}
}
// SamplingRefreshInterval creates a SamplerOption that sets how often the
// sampler will poll local agent for the appropriate sampling strategy.
func (SamplerOptionsFactory) SamplingRefreshInterval(samplingRefreshInterval time.Duration) SamplerOption {
return func(o *samplerOptions) {
o.samplingRefreshInterval = samplingRefreshInterval
}
}
// SamplingStrategyFetcher creates a SamplerOption that initializes sampling strategy fetcher.
func (SamplerOptionsFactory) SamplingStrategyFetcher(fetcher SamplingStrategyFetcher) SamplerOption {
return func(o *samplerOptions) {
o.samplingFetcher = fetcher
}
}
// SamplingStrategyParser creates a SamplerOption that initializes sampling strategy parser.
func (SamplerOptionsFactory) SamplingStrategyParser(parser SamplingStrategyParser) SamplerOption {
return func(o *samplerOptions) {
o.samplingParser = parser
}
}
// Updaters creates a SamplerOption that initializes sampler updaters.
func (SamplerOptionsFactory) Updaters(updaters ...SamplerUpdater) SamplerOption {
return func(o *samplerOptions) {
o.updaters = updaters
}
}
func (o *samplerOptions) applyOptionsAndDefaults(opts ...SamplerOption) *samplerOptions {
for _, option := range opts {
option(o)
}
if o.sampler == nil {
o.sampler = newProbabilisticSampler(0.001)
}
if o.logger == nil {
o.logger = log.NullLogger
}
if o.samplingServerURL == "" {
o.samplingServerURL = DefaultSamplingServerURL
}
if o.metrics == nil {
o.metrics = NewNullMetrics()
}
if o.samplingRefreshInterval <= 0 {
o.samplingRefreshInterval = defaultSamplingRefreshInterval
}
if o.samplingFetcher == nil {
o.samplingFetcher = &httpSamplingStrategyFetcher{
serverURL: o.samplingServerURL,
logger: o.logger,
}
}
if o.samplingParser == nil {
o.samplingParser = new(samplingStrategyParser)
}
if o.updaters == nil {
o.updaters = []SamplerUpdater{
&AdaptiveSamplerUpdater{
MaxOperations: o.posParams.MaxOperations,
OperationNameLateBinding: o.posParams.OperationNameLateBinding,
},
new(ProbabilisticSamplerUpdater),
new(RateLimitingSamplerUpdater),
}
}
return o
}
vendor/github.com/uber/jaeger-client-go/sampler_v2.go
Generated Vendored
+93
View File
@@ -0,0 +1,93 @@
// Copyright (c) 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.
package jaeger
// SamplingDecision is returned by the V2 samplers.
type SamplingDecision struct {
Sample bool
Retryable bool
Tags []Tag
}
// SamplerV2 is an extension of the V1 samplers that allows sampling decisions
// be made at different points of the span lifecycle.
type SamplerV2 interface {
OnCreateSpan(span *Span) SamplingDecision
OnSetOperationName(span *Span, operationName string) SamplingDecision
OnSetTag(span *Span, key string, value interface{}) SamplingDecision
OnFinishSpan(span *Span) SamplingDecision
// Close does a clean shutdown of the sampler, stopping any background
// go-routines it may have started.
Close()
}
// samplerV1toV2 wraps legacy V1 sampler into an adapter that make it look like V2.
func samplerV1toV2(s Sampler) SamplerV2 {
if s2, ok := s.(SamplerV2); ok {
return s2
}
type legacySamplerV1toV2Adapter struct {
legacySamplerV1Base
}
return &legacySamplerV1toV2Adapter{
legacySamplerV1Base: legacySamplerV1Base{
delegate: s.IsSampled,
},
}
}
// SamplerV2Base can be used by V2 samplers to implement dummy V1 methods.
// Supporting V1 API is required because Tracer configuration only accepts V1 Sampler
// for backwards compatibility reasons.
// TODO (breaking change) remove this in the next major release
type SamplerV2Base struct{}
// IsSampled implements IsSampled of Sampler.
func (SamplerV2Base) IsSampled(id TraceID, operation string) (sampled bool, tags []Tag) {
return false, nil
}
// Close implements Close of Sampler.
func (SamplerV2Base) Close() {}
// Equal implements Equal of Sampler.
func (SamplerV2Base) Equal(other Sampler) bool { return false }
// legacySamplerV1Base is used as a base for simple samplers that only implement
// the legacy isSampled() function that is not sensitive to its arguments.
type legacySamplerV1Base struct {
delegate func(id TraceID, operation string) (sampled bool, tags []Tag)
}
func (s *legacySamplerV1Base) OnCreateSpan(span *Span) SamplingDecision {
isSampled, tags := s.delegate(span.context.traceID, span.operationName)
return SamplingDecision{Sample: isSampled, Retryable: false, Tags: tags}
}
func (s *legacySamplerV1Base) OnSetOperationName(span *Span, operationName string) SamplingDecision {
isSampled, tags := s.delegate(span.context.traceID, span.operationName)
return SamplingDecision{Sample: isSampled, Retryable: false, Tags: tags}
}
func (s *legacySamplerV1Base) OnSetTag(span *Span, key string, value interface{}) SamplingDecision {
return SamplingDecision{Sample: false, Retryable: true}
}
func (s *legacySamplerV1Base) OnFinishSpan(span *Span) SamplingDecision {
return SamplingDecision{Sample: false, Retryable: true}
}
func (s *legacySamplerV1Base) Close() {}
vendor/github.com/uber/jaeger-client-go/span.go
Generated Vendored
+196 -25
View File
@@ -16,6 +16,7 @@ package jaeger
import (
"sync"
"sync/atomic"
"time"
"github.com/opentracing/opentracing-go"
@@ -25,10 +26,15 @@ import (
// Span implements opentracing.Span
type Span struct {
// referenceCounter used to increase the lifetime of
// the object before return it into the pool.
referenceCounter int32
sync.RWMutex
tracer *Tracer
// TODO: (breaking change) change to use a pointer
context SpanContext
// The name of the "operation" this span is an instance of.
@@ -60,18 +66,26 @@ type Span struct {
}
// Tag is a simple key value wrapper.
// TODO deprecate in the next major release, use opentracing.Tag instead.
// TODO (breaking change) deprecate in the next major release, use opentracing.Tag instead.
type Tag struct {
key string
value interface{}
}
// NewTag creates a new Tag.
// TODO (breaking change) deprecate in the next major release, use opentracing.Tag instead.
func NewTag(key string, value interface{}) Tag {
return Tag{key: key, value: value}
}
// SetOperationName sets or changes the operation name.
func (s *Span) SetOperationName(operationName string) opentracing.Span {
s.Lock()
defer s.Unlock()
if s.context.IsSampled() {
s.operationName = operationName
s.operationName = operationName
s.Unlock()
if !s.isSamplingFinalized() {
decision := s.tracer.sampler.OnSetOperationName(s, operationName)
s.applySamplingDecision(decision, true)
}
s.observer.OnSetOperationName(operationName)
return s
@@ -79,19 +93,85 @@ func (s *Span) SetOperationName(operationName string) opentracing.Span {
// SetTag implements SetTag() of opentracing.Span
func (s *Span) SetTag(key string, value interface{}) opentracing.Span {
return s.setTagInternal(key, value, true)
}
func (s *Span) setTagInternal(key string, value interface{}, lock bool) opentracing.Span {
s.observer.OnSetTag(key, value)
if key == string(ext.SamplingPriority) && !setSamplingPriority(s, value) {
return s
}
s.Lock()
defer s.Unlock()
if s.context.IsSampled() {
s.setTagNoLocking(key, value)
if !s.isSamplingFinalized() {
decision := s.tracer.sampler.OnSetTag(s, key, value)
s.applySamplingDecision(decision, lock)
}
if s.isWriteable() {
if lock {
s.Lock()
defer s.Unlock()
}
s.appendTagNoLocking(key, value)
}
return s
}
func (s *Span) setTagNoLocking(key string, value interface{}) {
// SpanContext returns span context
func (s *Span) SpanContext() SpanContext {
s.Lock()
defer s.Unlock()
return s.context
}
// StartTime returns span start time
func (s *Span) StartTime() time.Time {
s.Lock()
defer s.Unlock()
return s.startTime
}
// Duration returns span duration
func (s *Span) Duration() time.Duration {
s.Lock()
defer s.Unlock()
return s.duration
}
// Tags returns tags for span
func (s *Span) Tags() opentracing.Tags {
s.Lock()
defer s.Unlock()
var result = make(opentracing.Tags, len(s.tags))
for _, tag := range s.tags {
result[tag.key] = tag.value
}
return result
}
// Logs returns micro logs for span
func (s *Span) Logs() []opentracing.LogRecord {
s.Lock()
defer s.Unlock()
return append([]opentracing.LogRecord(nil), s.logs...)
}
// References returns references for this span
func (s *Span) References() []opentracing.SpanReference {
s.Lock()
defer s.Unlock()
if s.references == nil || len(s.references) == 0 {
return nil
}
result := make([]opentracing.SpanReference, len(s.references))
for i, r := range s.references {
result[i] = opentracing.SpanReference{Type: r.Type, ReferencedContext: r.Context}
}
return result
}
func (s *Span) appendTagNoLocking(key string, value interface{}) {
s.tags = append(s.tags, Tag{key: key, value: value})
}
@@ -111,7 +191,7 @@ func (s *Span) logFieldsNoLocking(fields ...log.Field) {
Fields: fields,
Timestamp: time.Now(),
}
s.appendLog(lr)
s.appendLogNoLocking(lr)
}
// LogKV implements opentracing.Span API
@@ -148,12 +228,12 @@ func (s *Span) Log(ld opentracing.LogData) {
if ld.Timestamp.IsZero() {
ld.Timestamp = s.tracer.timeNow()
}
s.appendLog(ld.ToLogRecord())
s.appendLogNoLocking(ld.ToLogRecord())
}
}
// this function should only be called while holding a Write lock
func (s *Span) appendLog(lr opentracing.LogRecord) {
func (s *Span) appendLogNoLocking(lr opentracing.LogRecord) {
// TODO add logic to limit number of logs per span (issue #46)
s.logs = append(s.logs, lr)
}
@@ -174,6 +254,8 @@ func (s *Span) BaggageItem(key string) string {
}
// Finish implements opentracing.Span API
// After finishing the Span object it returns back to the allocator unless the reporter retains it again,
// so after that, the Span object should no longer be used because it won't be valid anymore.
func (s *Span) Finish() {
s.FinishWithOptions(opentracing.FinishOptions{})
}
@@ -185,18 +267,27 @@ func (s *Span) FinishWithOptions(options opentracing.FinishOptions) {
}
s.observer.OnFinish(options)
s.Lock()
s.duration = options.FinishTime.Sub(s.startTime)
s.Unlock()
if !s.isSamplingFinalized() {
decision := s.tracer.sampler.OnFinishSpan(s)
s.applySamplingDecision(decision, true)
}
if s.context.IsSampled() {
s.duration = options.FinishTime.Sub(s.startTime)
// Note: bulk logs are not subject to maxLogsPerSpan limit
if options.LogRecords != nil {
s.logs = append(s.logs, options.LogRecords...)
}
for _, ld := range options.BulkLogData {
s.logs = append(s.logs, ld.ToLogRecord())
if len(options.LogRecords) > 0 || len(options.BulkLogData) > 0 {
s.Lock()
// Note: bulk logs are not subject to maxLogsPerSpan limit
if options.LogRecords != nil {
s.logs = append(s.logs, options.LogRecords...)
}
for _, ld := range options.BulkLogData {
s.logs = append(s.logs, ld.ToLogRecord())
}
s.Unlock()
}
}
s.Unlock()
// call reportSpan even for non-sampled traces, to return span to the pool
// and update metrics counter
s.tracer.reportSpan(s)
}
@@ -225,25 +316,105 @@ func (s *Span) OperationName() string {
return s.operationName
}
// Retain increases object counter to increase the lifetime of the object
func (s *Span) Retain() *Span {
atomic.AddInt32(&s.referenceCounter, 1)
return s
}
// Release decrements object counter and return to the
// allocator manager when counter will below zero
func (s *Span) Release() {
if atomic.AddInt32(&s.referenceCounter, -1) == -1 {
s.tracer.spanAllocator.Put(s)
}
}
// reset span state and release unused data
func (s *Span) reset() {
s.firstInProcess = false
s.context = emptyContext
s.operationName = ""
s.tracer = nil
s.startTime = time.Time{}
s.duration = 0
s.observer = nil
atomic.StoreInt32(&s.referenceCounter, 0)
// Note: To reuse memory we can save the pointers on the heap
s.tags = s.tags[:0]
s.logs = s.logs[:0]
s.references = s.references[:0]
}
func (s *Span) serviceName() string {
return s.tracer.serviceName
}
func (s *Span) applySamplingDecision(decision SamplingDecision, lock bool) {
if !decision.Retryable {
s.context.samplingState.setFinal()
}
if decision.Sample {
s.context.samplingState.setSampled()
if len(decision.Tags) > 0 {
if lock {
s.Lock()
defer s.Unlock()
}
for _, tag := range decision.Tags {
s.appendTagNoLocking(tag.key, tag.value)
}
}
}
}
// Span can be written to if it is sampled or the sampling decision has not been finalized.
func (s *Span) isWriteable() bool {
state := s.context.samplingState
return !state.isFinal() || state.isSampled()
}
func (s *Span) isSamplingFinalized() bool {
return s.context.samplingState.isFinal()
}
// setSamplingPriority returns true if the flag was updated successfully, false otherwise.
// The behavior of setSamplingPriority is surprising
// If noDebugFlagOnForcedSampling is set
// setSamplingPriority(span, 1) always sets only flagSampled
// If noDebugFlagOnForcedSampling is unset, and isDebugAllowed passes
// setSamplingPriority(span, 1) sets both flagSampled and flagDebug
// However,
// setSamplingPriority(span, 0) always only resets flagSampled
//
// This means that doing a setSamplingPriority(span, 1) followed by setSamplingPriority(span, 0) can
// leave flagDebug set
func setSamplingPriority(s *Span, value interface{}) bool {
s.Lock()
defer s.Unlock()
val, ok := value.(uint16)
if !ok {
return false
}
if val == 0 {
s.context.flags = s.context.flags & (^flagSampled)
s.context.samplingState.unsetSampled()
s.context.samplingState.setFinal()
return true
}
if s.tracer.isDebugAllowed(s.operationName) {
s.context.flags = s.context.flags | flagDebug | flagSampled
if s.tracer.options.noDebugFlagOnForcedSampling {
s.context.samplingState.setSampled()
s.context.samplingState.setFinal()
return true
} else if s.tracer.isDebugAllowed(s.operationName) {
s.context.samplingState.setDebugAndSampled()
s.context.samplingState.setFinal()
return true
}
return false
}
// EnableFirehose enables firehose flag on the span context
func EnableFirehose(s *Span) {
s.Lock()
defer s.Unlock()
s.context.samplingState.setFirehose()
}
vendor/github.com/uber/jaeger-client-go/span_allocator.go
Generated Vendored
+56
View File
@@ -0,0 +1,56 @@
// Copyright (c) 2019 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 jaeger
import "sync"
// SpanAllocator abstraction of managign span allocations
type SpanAllocator interface {
Get() *Span
Put(*Span)
}
type syncPollSpanAllocator struct {
spanPool sync.Pool
}
func newSyncPollSpanAllocator() SpanAllocator {
return &syncPollSpanAllocator{
spanPool: sync.Pool{New: func() interface{} {
return &Span{}
}},
}
}
func (pool *syncPollSpanAllocator) Get() *Span {
return pool.spanPool.Get().(*Span)
}
func (pool *syncPollSpanAllocator) Put(span *Span) {
span.reset()
pool.spanPool.Put(span)
}
type simpleSpanAllocator struct{}
func (pool simpleSpanAllocator) Get() *Span {
return &Span{}
}
func (pool simpleSpanAllocator) Put(span *Span) {
// @comment https://github.com/jaegertracing/jaeger-client-go/pull/381#issuecomment-475904351
// since finished spans are not reused, no need to reset them
// span.reset()
}
vendor/github.com/uber/jaeger-client-go/context.go → vendor/github.com/uber/jaeger-client-go/span_context.go
Generated Vendored
+144 -19
View File
@@ -19,11 +19,15 @@ import (
"fmt"
"strconv"
"strings"
"sync"
"go.uber.org/atomic"
)
const (
flagSampled = byte(1)
flagDebug = byte(2)
flagSampled = 1
flagDebug = 2
flagFirehose = 8
)
var (
@@ -55,9 +59,6 @@ type SpanContext struct {
// Should be 0 if the current span is a root span.
parentID SpanID
// flags is a bitmap containing such bits as 'sampled' and 'debug'.
flags byte
// Distributed Context baggage. The is a snapshot in time.
baggage map[string]string
@@ -66,6 +67,102 @@ type SpanContext struct {
//
// See JaegerDebugHeader in constants.go
debugID string
// samplingState is shared across all spans
samplingState *samplingState
// remote indicates that span context represents a remote parent
remote bool
}
type samplingState struct {
// Span context's state flags that are propagated across processes. Only lower 8 bits are used.
// We use an int32 instead of byte to be able to use CAS operations.
stateFlags atomic.Int32
// When state is not final, sampling will be retried on other span write operations,
// like SetOperationName / SetTag, and the spans will remain writable.
final atomic.Bool
// localRootSpan stores the SpanID of the first span created in this process for a given trace.
localRootSpan SpanID
// extendedState allows samplers to keep intermediate state.
// The keys and values in this map are completely opaque: interface{} -> interface{}.
extendedState sync.Map
}
func (s *samplingState) isLocalRootSpan(id SpanID) bool {
return id == s.localRootSpan
}
func (s *samplingState) setFlag(newFlag int32) {
swapped := false
for !swapped {
old := s.stateFlags.Load()
swapped = s.stateFlags.CAS(old, old|newFlag)
}
}
func (s *samplingState) unsetFlag(newFlag int32) {
swapped := false
for !swapped {
old := s.stateFlags.Load()
swapped = s.stateFlags.CAS(old, old&^newFlag)
}
}
func (s *samplingState) setSampled() {
s.setFlag(flagSampled)
}
func (s *samplingState) unsetSampled() {
s.unsetFlag(flagSampled)
}
func (s *samplingState) setDebugAndSampled() {
s.setFlag(flagDebug | flagSampled)
}
func (s *samplingState) setFirehose() {
s.setFlag(flagFirehose)
}
func (s *samplingState) setFlags(flags byte) {
s.stateFlags.Store(int32(flags))
}
func (s *samplingState) setFinal() {
s.final.Store(true)
}
func (s *samplingState) flags() byte {
return byte(s.stateFlags.Load())
}
func (s *samplingState) isSampled() bool {
return s.stateFlags.Load()&flagSampled == flagSampled
}
func (s *samplingState) isDebug() bool {
return s.stateFlags.Load()&flagDebug == flagDebug
}
func (s *samplingState) isFirehose() bool {
return s.stateFlags.Load()&flagFirehose == flagFirehose
}
func (s *samplingState) isFinal() bool {
return s.final.Load()
}
func (s *samplingState) extendedStateForKey(key interface{}, initValue func() interface{}) interface{} {
if value, ok := s.extendedState.Load(key); ok {
return value
}
value := initValue()
value, _ = s.extendedState.LoadOrStore(key, value)
return value
}
// ForeachBaggageItem implements ForeachBaggageItem() of opentracing.SpanContext
@@ -80,12 +177,28 @@ func (c SpanContext) ForeachBaggageItem(handler func(k, v string) bool) {
// IsSampled returns whether this trace was chosen for permanent storage
// by the sampling mechanism of the tracer.
func (c SpanContext) IsSampled() bool {
return (c.flags & flagSampled) == flagSampled
return c.samplingState.isSampled()
}
// IsDebug indicates whether sampling was explicitly requested by the service.
func (c SpanContext) IsDebug() bool {
return (c.flags & flagDebug) == flagDebug
return c.samplingState.isDebug()
}
// IsSamplingFinalized indicates whether the sampling decision has been finalized.
func (c SpanContext) IsSamplingFinalized() bool {
return c.samplingState.isFinal()
}
// IsFirehose indicates whether the firehose flag was set
func (c SpanContext) IsFirehose() bool {
return c.samplingState.isFirehose()
}
// ExtendedSamplingState returns the custom state object for a given key. If the value for this key does not exist,
// it is initialized via initValue function. This state can be used by samplers (e.g. x.PrioritySampler).
func (c SpanContext) ExtendedSamplingState(key interface{}, initValue func() interface{}) interface{} {
return c.samplingState.extendedStateForKey(key, initValue)
}
// IsValid indicates whether this context actually represents a valid trace.
@@ -93,11 +206,16 @@ func (c SpanContext) IsValid() bool {
return c.traceID.IsValid() && c.spanID != 0
}
// SetFirehose enables firehose mode for this trace.
func (c SpanContext) SetFirehose() {
c.samplingState.setFirehose()
}
func (c SpanContext) String() string {
if c.traceID.High == 0 {
return fmt.Sprintf("%x:%x:%x:%x", c.traceID.Low, uint64(c.spanID), uint64(c.parentID), c.flags)
return fmt.Sprintf("%x:%x:%x:%x", c.traceID.Low, uint64(c.spanID), uint64(c.parentID), c.samplingState.stateFlags.Load())
}
return fmt.Sprintf("%x%016x:%x:%x:%x", c.traceID.High, c.traceID.Low, uint64(c.spanID), uint64(c.parentID), c.flags)
return fmt.Sprintf("%x%016x:%x:%x:%x", c.traceID.High, c.traceID.Low, uint64(c.spanID), uint64(c.parentID), c.samplingState.stateFlags.Load())
}
// ContextFromString reconstructs the Context encoded in a string
@@ -124,7 +242,8 @@ func ContextFromString(value string) (SpanContext, error) {
if err != nil {
return emptyContext, err
}
context.flags = byte(flags)
context.samplingState = &samplingState{}
context.samplingState.setFlags(byte(flags))
return context, nil
}
@@ -143,18 +262,24 @@ func (c SpanContext) ParentID() SpanID {
return c.parentID
}
// Flags returns the bitmap containing such bits as 'sampled' and 'debug'.
func (c SpanContext) Flags() byte {
return c.samplingState.flags()
}
// NewSpanContext creates a new instance of SpanContext
func NewSpanContext(traceID TraceID, spanID, parentID SpanID, sampled bool, baggage map[string]string) SpanContext {
flags := byte(0)
samplingState := &samplingState{}
if sampled {
flags = flagSampled
samplingState.setSampled()
}
return SpanContext{
traceID: traceID,
spanID: spanID,
parentID: parentID,
flags: flags,
baggage: baggage}
traceID: traceID,
spanID: spanID,
parentID: parentID,
samplingState: samplingState,
baggage: baggage}
}
// CopyFrom copies data from ctx into this context, including span identity and baggage.
@@ -163,7 +288,7 @@ func (c *SpanContext) CopyFrom(ctx *SpanContext) {
c.traceID = ctx.traceID
c.spanID = ctx.spanID
c.parentID = ctx.parentID
c.flags = ctx.flags
c.samplingState = ctx.samplingState
if l := len(ctx.baggage); l > 0 {
c.baggage = make(map[string]string, l)
for k, v := range ctx.baggage {
@@ -187,7 +312,7 @@ func (c SpanContext) WithBaggageItem(key, value string) SpanContext {
newBaggage[key] = value
}
// Use positional parameters so the compiler will help catch new fields.
return SpanContext{c.traceID, c.spanID, c.parentID, c.flags, newBaggage, ""}
return SpanContext{c.traceID, c.spanID, c.parentID, newBaggage, "", c.samplingState, c.remote}
}
// isDebugIDContainerOnly returns true when the instance of the context is only
vendor/github.com/uber/jaeger-client-go/tracer.go
Generated Vendored
+155 -114
View File
@@ -38,7 +38,7 @@ type Tracer struct {
serviceName string
hostIPv4 uint32 // this is for zipkin endpoint conversion
sampler Sampler
sampler SamplerV2
reporter Reporter
metrics Metrics
logger log.Logger
@@ -47,15 +47,15 @@ type Tracer struct {
randomNumber func() uint64
options struct {
poolSpans bool
gen128Bit bool // whether to generate 128bit trace IDs
zipkinSharedRPCSpan bool
highTraceIDGenerator func() uint64 // custom high trace ID generator
maxTagValueLength int
gen128Bit bool // whether to generate 128bit trace IDs
zipkinSharedRPCSpan bool
highTraceIDGenerator func() uint64 // custom high trace ID generator
maxTagValueLength int
noDebugFlagOnForcedSampling bool
// more options to come
}
// pool for Span objects
spanPool sync.Pool
// allocator of Span objects
spanAllocator SpanAllocator
injectors map[interface{}]Injector
extractors map[interface{}]Extractor
@@ -74,6 +74,7 @@ type Tracer struct {
// NewTracer creates Tracer implementation that reports tracing to Jaeger.
// The returned io.Closer can be used in shutdown hooks to ensure that the internal
// queue of the Reporter is drained and all buffered spans are submitted to collectors.
// TODO (breaking change) return *Tracer only, without closer.
func NewTracer(
serviceName string,
sampler Sampler,
@@ -81,15 +82,13 @@ func NewTracer(
options ...TracerOption,
) (opentracing.Tracer, io.Closer) {
t := &Tracer{
serviceName: serviceName,
sampler: sampler,
reporter: reporter,
injectors: make(map[interface{}]Injector),
extractors: make(map[interface{}]Extractor),
metrics: *NewNullMetrics(),
spanPool: sync.Pool{New: func() interface{} {
return &Span{}
}},
serviceName: serviceName,
sampler: samplerV1toV2(sampler),
reporter: reporter,
injectors: make(map[interface{}]Injector),
extractors: make(map[interface{}]Extractor),
metrics: *NewNullMetrics(),
spanAllocator: simpleSpanAllocator{},
}
for _, option := range options {
@@ -148,7 +147,15 @@ func NewTracer(
if hostname, err := os.Hostname(); err == nil {
t.tags = append(t.tags, Tag{key: TracerHostnameTagKey, value: hostname})
}
if ip, err := utils.HostIP(); err == nil {
if ipval, ok := t.getTag(TracerIPTagKey); ok {
ipv4, err := utils.ParseIPToUint32(ipval.(string))
if err != nil {
t.hostIPv4 = 0
t.logger.Error("Unable to convert the externally provided ip to uint32: " + err.Error())
} else {
t.hostIPv4 = ipv4
}
} else if ip, err := utils.HostIP(); err == nil {
t.tags = append(t.tags, Tag{key: TracerIPTagKey, value: ip.String()})
t.hostIPv4 = utils.PackIPAsUint32(ip)
} else {
@@ -217,20 +224,30 @@ func (t *Tracer) startSpanWithOptions(
var references []Reference
var parent SpanContext
var hasParent bool // need this because `parent` is a value, not reference
var ctx SpanContext
var isSelfRef bool
for _, ref := range options.References {
ctx, ok := ref.ReferencedContext.(SpanContext)
ctxRef, ok := ref.ReferencedContext.(SpanContext)
if !ok {
t.logger.Error(fmt.Sprintf(
"Reference contains invalid type of SpanReference: %s",
reflect.ValueOf(ref.ReferencedContext)))
continue
}
if !isValidReference(ctx) {
if !isValidReference(ctxRef) {
continue
}
references = append(references, Reference{Type: ref.Type, Context: ctx})
if ref.Type == selfRefType {
isSelfRef = true
ctx = ctxRef
continue
}
references = append(references, Reference{Type: ref.Type, Context: ctxRef})
if !hasParent {
parent = ctx
parent = ctxRef
hasParent = ref.Type == opentracing.ChildOfRef
}
}
@@ -245,60 +262,77 @@ func (t *Tracer) startSpanWithOptions(
rpcServer = (v == ext.SpanKindRPCServerEnum || v == string(ext.SpanKindRPCServerEnum))
}
var samplerTags []Tag
var ctx SpanContext
var internalTags []Tag
newTrace := false
if !hasParent || !parent.IsValid() {
newTrace = true
ctx.traceID.Low = t.randomID()
if t.options.gen128Bit {
ctx.traceID.High = t.options.highTraceIDGenerator()
}
ctx.spanID = SpanID(ctx.traceID.Low)
ctx.parentID = 0
ctx.flags = byte(0)
if hasParent && parent.isDebugIDContainerOnly() && t.isDebugAllowed(operationName) {
ctx.flags |= (flagSampled | flagDebug)
samplerTags = []Tag{{key: JaegerDebugHeader, value: parent.debugID}}
} else if sampled, tags := t.sampler.IsSampled(ctx.traceID, operationName); sampled {
ctx.flags |= flagSampled
samplerTags = tags
}
} else {
ctx.traceID = parent.traceID
if rpcServer && t.options.zipkinSharedRPCSpan {
// Support Zipkin's one-span-per-RPC model
ctx.spanID = parent.spanID
ctx.parentID = parent.parentID
if !isSelfRef {
if !hasParent || !parent.IsValid() {
newTrace = true
ctx.traceID.Low = t.randomID()
if t.options.gen128Bit {
ctx.traceID.High = t.options.highTraceIDGenerator()
}
ctx.spanID = SpanID(ctx.traceID.Low)
ctx.parentID = 0
ctx.samplingState = &samplingState{
localRootSpan: ctx.spanID,
}
if hasParent && parent.isDebugIDContainerOnly() && t.isDebugAllowed(operationName) {
ctx.samplingState.setDebugAndSampled()
internalTags = append(internalTags, Tag{key: JaegerDebugHeader, value: parent.debugID})
}
} else {
ctx.spanID = SpanID(t.randomID())
ctx.parentID = parent.spanID
ctx.traceID = parent.traceID
if rpcServer && t.options.zipkinSharedRPCSpan {
// Support Zipkin's one-span-per-RPC model
ctx.spanID = parent.spanID
ctx.parentID = parent.parentID
} else {
ctx.spanID = SpanID(t.randomID())
ctx.parentID = parent.spanID
}
ctx.samplingState = parent.samplingState
if parent.remote {
ctx.samplingState.setFinal()
ctx.samplingState.localRootSpan = ctx.spanID
}
}
ctx.flags = parent.flags
}
if hasParent {
// copy baggage items
if l := len(parent.baggage); l > 0 {
ctx.baggage = make(map[string]string, len(parent.baggage))
for k, v := range parent.baggage {
ctx.baggage[k] = v
if hasParent {
// copy baggage items
if l := len(parent.baggage); l > 0 {
ctx.baggage = make(map[string]string, len(parent.baggage))
for k, v := range parent.baggage {
ctx.baggage[k] = v
}
}
}
}
sp := t.newSpan()
sp.context = ctx
sp.tracer = t
sp.operationName = operationName
sp.startTime = options.StartTime
sp.duration = 0
sp.references = references
sp.firstInProcess = rpcServer || sp.context.parentID == 0
if !sp.isSamplingFinalized() {
decision := t.sampler.OnCreateSpan(sp)
sp.applySamplingDecision(decision, false)
}
sp.observer = t.observer.OnStartSpan(sp, operationName, options)
return t.startSpanInternal(
sp,
operationName,
options.StartTime,
samplerTags,
options.Tags,
newTrace,
rpcServer,
references,
)
if tagsTotalLength := len(options.Tags) + len(internalTags); tagsTotalLength > 0 {
if sp.tags == nil || cap(sp.tags) < tagsTotalLength {
sp.tags = make([]Tag, 0, tagsTotalLength)
}
sp.tags = append(sp.tags, internalTags...)
for k, v := range options.Tags {
sp.setTagInternal(k, v, false)
}
}
t.emitNewSpanMetrics(sp, newTrace)
return sp
}
// Inject implements Inject() method of opentracing.Tracer
@@ -323,6 +357,7 @@ func (t *Tracer) Extract(
if err != nil {
return nil, err // ensure returned spanCtx is nil
}
spanCtx.remote = true
return spanCtx, nil
}
return nil, opentracing.ErrUnsupportedFormat
@@ -333,10 +368,10 @@ func (t *Tracer) Close() error {
t.reporter.Close()
t.sampler.Close()
if mgr, ok := t.baggageRestrictionManager.(io.Closer); ok {
mgr.Close()
_ = mgr.Close()
}
if throttler, ok := t.debugThrottler.(io.Closer); ok {
throttler.Close()
_ = throttler.Close()
}
return nil
}
@@ -350,55 +385,38 @@ func (t *Tracer) Tags() []opentracing.Tag {
return tags
}
// getTag returns the value of specific tag, if not exists, return nil.
// TODO only used by tests, move there.
func (t *Tracer) getTag(key string) (interface{}, bool) {
for _, tag := range t.tags {
if tag.key == key {
return tag.value, true
}
}
return nil, false
}
// newSpan returns an instance of a clean Span object.
// If options.PoolSpans is true, the spans are retrieved from an object pool.
func (t *Tracer) newSpan() *Span {
if !t.options.poolSpans {
return &Span{}
}
sp := t.spanPool.Get().(*Span)
sp.context = emptyContext
sp.tracer = nil
sp.tags = nil
sp.logs = nil
return sp
return t.spanAllocator.Get()
}
func (t *Tracer) startSpanInternal(
sp *Span,
operationName string,
startTime time.Time,
internalTags []Tag,
tags opentracing.Tags,
newTrace bool,
rpcServer bool,
references []Reference,
) *Span {
sp.tracer = t
sp.operationName = operationName
sp.startTime = startTime
sp.duration = 0
sp.references = references
sp.firstInProcess = rpcServer || sp.context.parentID == 0
if len(tags) > 0 || len(internalTags) > 0 {
sp.tags = make([]Tag, len(internalTags), len(tags)+len(internalTags))
copy(sp.tags, internalTags)
for k, v := range tags {
sp.observer.OnSetTag(k, v)
if k == string(ext.SamplingPriority) && !setSamplingPriority(sp, v) {
continue
}
sp.setTagNoLocking(k, v)
// emitNewSpanMetrics generates metrics on the number of started spans and traces.
// newTrace param: we cannot simply check for parentID==0 because in Zipkin model the
// server-side RPC span has the exact same trace/span/parent IDs as the
// calling client-side span, but obviously the server side span is
// no longer a root span of the trace.
func (t *Tracer) emitNewSpanMetrics(sp *Span, newTrace bool) {
if !sp.isSamplingFinalized() {
t.metrics.SpansStartedDelayedSampling.Inc(1)
if newTrace {
t.metrics.TracesStartedDelayedSampling.Inc(1)
}
}
// emit metrics
if sp.context.IsSampled() {
// joining a trace is not possible, because sampling decision inherited from upstream is final
} else if sp.context.IsSampled() {
t.metrics.SpansStartedSampled.Inc(1)
if newTrace {
// We cannot simply check for parentID==0 because in Zipkin model the
// server-side RPC span has the exact same trace/span/parent IDs as the
// calling client-side span, but obviously the server side span is
// no longer a root span of the trace.
t.metrics.TracesStartedSampled.Inc(1)
} else if sp.firstInProcess {
t.metrics.TracesJoinedSampled.Inc(1)
@@ -411,17 +429,25 @@ func (t *Tracer) startSpanInternal(
t.metrics.TracesJoinedNotSampled.Inc(1)
}
}
return sp
}
func (t *Tracer) reportSpan(sp *Span) {
t.metrics.SpansFinished.Inc(1)
if !sp.isSamplingFinalized() {
t.metrics.SpansFinishedDelayedSampling.Inc(1)
} else if sp.context.IsSampled() {
t.metrics.SpansFinishedSampled.Inc(1)
} else {
t.metrics.SpansFinishedNotSampled.Inc(1)
}
// Note: if the reporter is processing Span asynchronously then it needs to Retain() the span,
// and then Release() it when no longer needed.
// Otherwise, the span may be reused for another trace and its data may be overwritten.
if sp.context.IsSampled() {
t.reporter.Report(sp)
}
if t.options.poolSpans {
t.spanPool.Put(sp)
}
sp.Release()
}
// randomID generates a random trace/span ID, using tracer.random() generator.
@@ -443,3 +469,18 @@ func (t *Tracer) setBaggage(sp *Span, key, value string) {
func (t *Tracer) isDebugAllowed(operation string) bool {
return t.debugThrottler.IsAllowed(operation)
}
// Sampler returns the sampler given to the tracer at creation.
func (t *Tracer) Sampler() SamplerV2 {
return t.sampler
}
// SelfRef creates an opentracing compliant SpanReference from a jaeger
// SpanContext. This is a factory function in order to encapsulate jaeger specific
// types.
func SelfRef(ctx SpanContext) opentracing.SpanReference {
return opentracing.SpanReference{
Type: selfRefType,
ReferencedContext: ctx,
}
}
vendor/github.com/uber/jaeger-client-go/tracer_options.go
Generated Vendored
+11 -1
View File
@@ -81,7 +81,11 @@ func (tracerOptions) RandomNumber(randomNumber func() uint64) TracerOption {
// that can access parent spans after those spans have been finished.
func (tracerOptions) PoolSpans(poolSpans bool) TracerOption {
return func(tracer *Tracer) {
tracer.options.poolSpans = poolSpans
if poolSpans {
tracer.spanAllocator = newSyncPollSpanAllocator()
} else {
tracer.spanAllocator = simpleSpanAllocator{}
}
}
}
@@ -122,6 +126,12 @@ func (tracerOptions) Gen128Bit(gen128Bit bool) TracerOption {
}
}
func (tracerOptions) NoDebugFlagOnForcedSampling(noDebugFlagOnForcedSampling bool) TracerOption {
return func(tracer *Tracer) {
tracer.options.noDebugFlagOnForcedSampling = noDebugFlagOnForcedSampling
}
}
func (tracerOptions) HighTraceIDGenerator(highTraceIDGenerator func() uint64) TracerOption {
return func(tracer *Tracer) {
tracer.options.highTraceIDGenerator = highTraceIDGenerator
vendor/github.com/uber/jaeger-client-go/utils/rate_limiter.go
Generated Vendored
+64 -29
View File
@@ -20,22 +20,15 @@ import (
)
// RateLimiter is a filter used to check if a message that is worth itemCost units is within the rate limits.
//
// TODO (breaking change) remove this interface in favor of public struct below
//
// Deprecated, use ReconfigurableRateLimiter.
type RateLimiter interface {
CheckCredit(itemCost float64) bool
}
type rateLimiter struct {
sync.Mutex
creditsPerSecond float64
balance float64
maxBalance float64
lastTick time.Time
timeNow func() time.Time
}
// NewRateLimiter creates a new rate limiter based on leaky bucket algorithm, formulated in terms of a
// ReconfigurableRateLimiter is a rate limiter based on leaky bucket algorithm, formulated in terms of a
// credits balance that is replenished every time CheckCredit() method is called (tick) by the amount proportional
// to the time elapsed since the last tick, up to max of creditsPerSecond. A call to CheckCredit() takes a cost
// of an item we want to pay with the balance. If the balance exceeds the cost of the item, the item is "purchased"
@@ -47,31 +40,73 @@ type rateLimiter struct {
//
// It can also be used to limit the rate of traffic in bytes, by setting creditsPerSecond to desired throughput
// as bytes/second, and calling CheckCredit() with the actual message size.
func NewRateLimiter(creditsPerSecond, maxBalance float64) RateLimiter {
return &rateLimiter{
//
// TODO (breaking change) rename to RateLimiter once the interface is removed
type ReconfigurableRateLimiter struct {
lock sync.Mutex
creditsPerSecond float64
balance float64
maxBalance float64
lastTick time.Time
timeNow func() time.Time
}
// NewRateLimiter creates a new ReconfigurableRateLimiter.
func NewRateLimiter(creditsPerSecond, maxBalance float64) *ReconfigurableRateLimiter {
return &ReconfigurableRateLimiter{
creditsPerSecond: creditsPerSecond,
balance: maxBalance,
maxBalance: maxBalance,
lastTick: time.Now(),
timeNow: time.Now}
timeNow: time.Now,
}
}
func (b *rateLimiter) CheckCredit(itemCost float64) bool {
b.Lock()
defer b.Unlock()
// calculate how much time passed since the last tick, and update current tick
currentTime := b.timeNow()
elapsedTime := currentTime.Sub(b.lastTick)
b.lastTick = currentTime
// calculate how much credit have we accumulated since the last tick
b.balance += elapsedTime.Seconds() * b.creditsPerSecond
if b.balance > b.maxBalance {
b.balance = b.maxBalance
}
// CheckCredit tries to reduce the current balance by itemCost provided that the current balance
// is not lest than itemCost.
func (rl *ReconfigurableRateLimiter) CheckCredit(itemCost float64) bool {
rl.lock.Lock()
defer rl.lock.Unlock()
// if we have enough credits to pay for current item, then reduce balance and allow
if b.balance >= itemCost {
b.balance -= itemCost
if rl.balance >= itemCost {
rl.balance -= itemCost
return true
}
// otherwise check if balance can be increased due to time elapsed, and try again
rl.updateBalance()
if rl.balance >= itemCost {
rl.balance -= itemCost
return true
}
return false
}
// updateBalance recalculates current balance based on time elapsed. Must be called while holding a lock.
func (rl *ReconfigurableRateLimiter) updateBalance() {
// calculate how much time passed since the last tick, and update current tick
currentTime := rl.timeNow()
elapsedTime := currentTime.Sub(rl.lastTick)
rl.lastTick = currentTime
// calculate how much credit have we accumulated since the last tick
rl.balance += elapsedTime.Seconds() * rl.creditsPerSecond
if rl.balance > rl.maxBalance {
rl.balance = rl.maxBalance
}
}
// Update changes the main parameters of the rate limiter in-place, while retaining
// the current accumulated balance (pro-rated to the new maxBalance value). Using this method
// instead of creating a new rate limiter helps to avoid thundering herd when sampling
// strategies are updated.
func (rl *ReconfigurableRateLimiter) Update(creditsPerSecond, maxBalance float64) {
rl.lock.Lock()
defer rl.lock.Unlock()
rl.updateBalance() // get up to date balance
rl.balance = rl.balance * maxBalance / rl.maxBalance
rl.creditsPerSecond = creditsPerSecond
rl.maxBalance = maxBalance
}
vendor/github.com/uber/jaeger-client-go/zipkin.go
Generated Vendored
+3 -2
View File
@@ -55,7 +55,7 @@ func (p *zipkinPropagator) Inject(
carrier.SetTraceID(ctx.TraceID().Low) // TODO this cannot work with 128bit IDs
carrier.SetSpanID(uint64(ctx.SpanID()))
carrier.SetParentID(uint64(ctx.ParentID()))
carrier.SetFlags(ctx.flags)
carrier.SetFlags(ctx.samplingState.flags())
return nil
}
@@ -71,6 +71,7 @@ func (p *zipkinPropagator) Extract(abstractCarrier interface{}) (SpanContext, er
ctx.traceID.Low = carrier.TraceID()
ctx.spanID = SpanID(carrier.SpanID())
ctx.parentID = SpanID(carrier.ParentID())
ctx.flags = carrier.Flags()
ctx.samplingState = &samplingState{}
ctx.samplingState.setFlags(carrier.Flags())
return ctx, nil
}
vendor/github.com/uber/jaeger-lib/metrics/metrics.go
Generated Vendored
+1 -1
View File
@@ -37,7 +37,7 @@ func MustInit(metrics interface{}, factory Factory, globalTags map[string]string
}
}
// Init does the same as Init, but returns an error instead of
// Init does the same as MustInit, but returns an error instead of
// panicking.
func Init(m interface{}, factory Factory, globalTags map[string]string) error {
// Allow user to opt out of reporting metrics by passing in nil.