Add OTel tracing bucket

CHANGELOG.md

@@ -13,7 +13,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 ### Fixed
 - [#33](https://github.com/thanos-io/objstore/pull/33) Tracing: Add `ContextWithTracer()` to inject the tracer into the context.
 - [#34](https://github.com/thanos-io/objstore/pull/34) Fix ignored options when creating shared credential Azure client.
-- [#62](https://github.com/thanos-io/objstore/pull/62) S3: Fix ignored context cancellation in `Iter` method. 
+- [#62](https://github.com/thanos-io/objstore/pull/62) S3: Fix ignored context cancellation in `Iter` method.
 
 ### Added
 - [#15](https://github.com/thanos-io/objstore/pull/15) Add Oracle Cloud Infrastructure Object Storage Bucket support.
@@ -23,7 +23,10 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#43](https://github.com/thanos-io/objstore/pull/43) filesystem: abort filesystem bucket operations if the context has been cancelled
 - [#44](https://github.com/thanos-io/objstore/pull/44) Add new metric to count total number of fetched bytes from bucket
 - [#50](https://github.com/thanos-io/objstore/pull/50) Add Huawei Cloud OBS Object Storage Support
-- [#59](https://github.com/thanos-io/objstore/pull/59) Adding method `IsCustomerManagedKeyError` on the bucket interface. 
+- [#59](https://github.com/thanos-io/objstore/pull/59) Adding method `IsCustomerManagedKeyError` on the bucket interface.
+- [#61](https://github.com/thanos-io/objstore/pull/61) Add OpenTelemetry TracingBucket.
+    > This also changes the behaviour of `client.NewBucket`. Now it returns, uninstrumented and untraced bucket.
+    You can combine `objstore.WrapWithMetrics` and `tracing/{opentelemetry,opentracing}.WrapWithTraces` to have old behavior.
 
 ### Changed
 - [#38](https://github.com/thanos-io/objstore/pull/38) *: Upgrade minio-go version to `v7.0.45`.

README.md

@@ -60,7 +60,7 @@ type Bucket interface {
 	Upload(ctx context.Context, name string, r io.Reader) error
 
 	// Delete removes the object with the given name.
-	// If object does not exists in the moment of deletion, Delete should throw error.
+	// If object does not exist in the moment of deletion, Delete should throw error.
 	Delete(ctx context.Context, name string) error
 
 ```
@@ -88,7 +88,7 @@ type BucketReader interface {
 	// IsObjNotFoundErr returns true if error means that object is not found. Relevant to Get operations.
 	IsObjNotFoundErr(err error) bool
 
-	// Attributes returns information about the specified object.
+	// IsCustomerManagedKeyError returns true if the permissions for key used to encrypt the object was revoked.
 ```
 
 Those interfaces represent the object storage operations your code can use from `objstore` clients.
@@ -128,7 +128,7 @@ Current object storage client implementations:
 | [Baidu BOS](#baidu-bos)                                                                   | Beta               | Production Usage      | no                | @yahaa                           |
 | [Local Filesystem](#filesystem)                                                           | Stable             | Testing and Demo only | yes               | @bwplotka                        |
 | [Oracle Cloud Infrastructure Object Storage](#oracle-cloud-infrastructure-object-storage) | Beta               | Production Usage      | yes               | @aarontams,@gaurav-05,@ericrrath |
-| [HuaweiCloud OBS](#huaweicloud-obs)                                                       | Beta               | Production Usage      | no               | @setoru                          |
+| [HuaweiCloud OBS](#huaweicloud-obs)                                                       | Beta               | Production Usage      | no                | @setoru                          |
 
 **Missing support to some object storage?** Check out [how to add your client section](#how-to-add-a-new-client-to-thanos)
 
@@ -153,6 +153,7 @@ config:
   insecure: false
   signature_version2: false
   secret_key: ""
+  session_token: ""
   put_user_metadata: {}
   http_config:
     idle_conn_timeout: 1m30s
@@ -643,17 +644,16 @@ You can also include any of the optional configuration just like the example in
 
 ##### HuaweiCloud OBS
 
-To use HuaweiCloud OBS as an object store, you should apply for a HuaweiCloud Account to create an object storage bucket at first.
-More details: [HuaweiCloud OBS](https://support.huaweicloud.com/obs/index.html)
+To use HuaweiCloud OBS as an object store, you should apply for a HuaweiCloud Account to create an object storage bucket at first. More details: [HuaweiCloud OBS](https://support.huaweicloud.com/obs/index.html)
 
 To configure HuaweiCloud Account to use OBS as storage store you need to set these parameters in YAML format stored in a file:
 
-```yaml mdox-exec="go run scripts/cfggen/main.go --name=cos.Config"
-type: OBS                    
-config:                      
-  bucket: ""                 
-  endpoint: ""               
-  access_key: ""             
+```yaml mdox-exec="go run scripts/cfggen/main.go --name=obs.Config"
+type: OBS
+config:
+  bucket: ""
+  endpoint: ""
+  access_key: ""
   secret_key: ""
   http_config:
     idle_conn_timeout: 1m30s
@@ -674,7 +674,7 @@ config:
 prefix: ""
 ```
 
-The `access_key` and `secret_key` field is required. The `http_config` field is optional for optimize HTTP transport settings. 
+The `access_key` and `secret_key` field is required. The `http_config` field is optional for optimize HTTP transport settings.
 
 #### How to add a new client to Thanos?
 

client/factory.go

@@ -23,7 +23,6 @@ import (
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/pkg/errors"
-	"github.com/prometheus/client_golang/prometheus"
 	"gopkg.in/yaml.v2"
 )
 
@@ -50,7 +49,7 @@ type BucketConfig struct {
 
 // NewBucket initializes and returns new object storage clients.
 // NOTE: confContentYaml can contain secrets.
-func NewBucket(logger log.Logger, confContentYaml []byte, reg prometheus.Registerer, component string) (objstore.InstrumentedBucket, error) {
+func NewBucket(logger log.Logger, confContentYaml []byte, component string) (objstore.Bucket, error) {
 	level.Info(logger).Log("msg", "loading bucket configuration")
 	bucketConf := &BucketConfig{}
 	if err := yaml.UnmarshalStrict(confContentYaml, bucketConf); err != nil {
@@ -91,5 +90,5 @@ func NewBucket(logger log.Logger, confContentYaml []byte, reg prometheus.Registe
 		return nil, errors.Wrap(err, fmt.Sprintf("create %s client", bucketConf.Type))
 	}
 
-	return objstore.NewTracingBucket(objstore.BucketWithMetrics(bucket.Name(), objstore.NewPrefixedBucket(bucket, bucketConf.Prefix), reg)), nil
+	return objstore.NewPrefixedBucket(bucket, bucketConf.Prefix), nil
 }

client/factory_test.go

@@ -0,0 +1,91 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package client
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+
+	"github.com/go-kit/log"
+	"go.opentelemetry.io/otel/trace"
+
+	"github.com/thanos-io/objstore/tracing/opentelemetry"
+	"github.com/thanos-io/objstore/tracing/opentracing"
+)
+
+func ExampleBucket() {
+	// Read the configuration file.
+	confContentYaml, err := ioutil.ReadFile("testconf/filesystem.conf.yml")
+	if err != nil {
+		panic(err)
+	}
+
+	// Create a new bucket.
+	bucket, err := NewBucket(log.NewNopLogger(), confContentYaml, "example")
+	if err != nil {
+		panic(err)
+	}
+
+	// Test it.
+	exists, err := bucket.Exists(context.Background(), "example")
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println(exists)
+	// Output:
+	// false
+}
+
+func ExampleTracingBucketUsingOpenTracing() { //nolint:govet
+	// Read the configuration file.
+	confContentYaml, err := ioutil.ReadFile("testconf/filesystem.conf.yml")
+	if err != nil {
+		panic(err)
+	}
+
+	// Create a new bucket.
+	bucket, err := NewBucket(log.NewNopLogger(), confContentYaml, "example")
+	if err != nil {
+		panic(err)
+	}
+
+	// Wrap it with tracing.
+	bucket = opentracing.WrapWithTraces(bucket)
+
+	// Test it.
+	exists, err := bucket.Exists(context.Background(), "example")
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println(exists)
+	// Output:
+	// false
+}
+
+func ExampleTracingBucketUsingOpenTelemetry() { //nolint:govet
+	// Read the configuration file.
+	confContentYaml, err := ioutil.ReadFile("testconf/filesystem.conf.yml")
+	if err != nil {
+		panic(err)
+	}
+
+	// Create a new bucket.
+	bucket, err := NewBucket(log.NewNopLogger(), confContentYaml, "example")
+	if err != nil {
+		panic(err)
+	}
+
+	// Wrap it with tracing.
+	bucket = opentelemetry.WrapWithTraces(bucket, trace.NewNoopTracerProvider().Tracer("bucket"))
+
+	// Test it.
+	exists, err := bucket.Exists(context.Background(), "example")
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println(exists)
+	// Output:
+	// false
+}

client/testconf/filesystem.conf.yml

@@ -0,0 +1,3 @@
+type: "FILESYSTEM"
+config:
+  directory: "./data"

go.mod

@@ -21,6 +21,8 @@ require (
 	github.com/prometheus/client_golang v1.12.2
 	github.com/prometheus/common v0.36.0
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.40
+	go.opentelemetry.io/otel v1.16.0
+	go.opentelemetry.io/otel/trace v1.16.0
 	go.uber.org/atomic v1.9.0
 	golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5
 	golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4
@@ -92,7 +94,6 @@ require (
 	google.golang.org/protobuf v1.28.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/ini.v1 v1.66.6 // indirect
-	gopkg.in/yaml.v3 v3.0.0 // indirect
 )
 
 require (

go.sum

@@ -223,8 +223,8 @@ github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
-github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
 github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck=
 github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD/fhyJ8=
 github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17icRSOU623lUBU=
@@ -385,7 +385,7 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
+github.com/stretchr/testify v1.8.3 h1:RP3t2pwF7cMEbC1dqtB6poj3niw/9gnV4Cjg5oW5gtY=
 github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common v1.0.194/go.mod h1:7sCQWVkxcsR38nffDW057DRGk8mUjK1Ing/EFOK8s8Y=
 github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/kms v1.0.194/go.mod h1:yrBKWhChnDqNz1xuXdSbWXG56XawEq0G5j1lg4VwBD4=
 github.com/tencentyun/cos-go-sdk-v5 v0.7.40 h1:W6vDGKCHe4wBACI1d2UgE6+50sJFhRWU4O8IB2ozzxM=
@@ -403,6 +403,10 @@ go.opencensus.io v0.22.4/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw=
 go.opencensus.io v0.22.5/go.mod h1:5pWMHQbX5EPX2/62yrJeAkowc+lfs/XD7Uxpq3pI6kk=
 go.opencensus.io v0.23.0 h1:gqCw0LfLxScz8irSi8exQc7fyQ0fKQU/qnC/X8+V/1M=
 go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
+go.opentelemetry.io/otel v1.16.0 h1:Z7GVAX/UkAXPKsy94IU+i6thsQS4nb7LviLpnaNeW8s=
+go.opentelemetry.io/otel v1.16.0/go.mod h1:vl0h9NUa1D5s1nv3A5vZOYWn8av4K8Ml6JDeHrT/bx4=
+go.opentelemetry.io/otel/trace v1.16.0 h1:8JRpaObFoW0pxuVPapkgH8UhHQj+bJW8jJsCZEu5MQs=
+go.opentelemetry.io/otel/trace v1.16.0/go.mod h1:Yt9vYq1SdNz3xdjZZK7wcXv1qv2pwLkqr2QVwea0ef0=
 go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI=
 go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE=
 go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
@@ -858,8 +862,7 @@ gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-gopkg.in/yaml.v3 v3.0.0 h1:hjy8E9ON/egN1tAYqKb61G10WtihqetD4sz2H+8nIeA=
-gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

objstore.go

@@ -400,9 +400,9 @@ type IsOpFailureExpectedFunc func(error) bool
 
 var _ InstrumentedBucket = &metricBucket{}
 
-// BucketWithMetrics takes a bucket and registers metrics with the given registry for
+// WrapWithMetrics takes a bucket and registers metrics with the given registry for
 // operations run against the bucket.
-func BucketWithMetrics(name string, b Bucket, reg prometheus.Registerer) *metricBucket {
+func WrapWithMetrics(b Bucket, reg prometheus.Registerer, name string) *metricBucket {
 	bkt := &metricBucket{
 		bkt:                 b,
 		isOpFailureExpected: func(err error) bool { return false },

objstore_test.go

@@ -20,7 +20,7 @@ import (
 )
 
 func TestMetricBucket_Close(t *testing.T) {
-	bkt := BucketWithMetrics("abc", NewInMemBucket(), nil)
+	bkt := WrapWithMetrics(NewInMemBucket(), nil, "abc")
 	// Expected initialized metrics.
 	testutil.Equals(t, 7, promtest.CollectAndCount(bkt.ops))
 	testutil.Equals(t, 7, promtest.CollectAndCount(bkt.opsFailures))
@@ -72,30 +72,9 @@ func TestMetricBucket_Close(t *testing.T) {
 	testutil.Assert(t, promtest.ToFloat64(bkt.lastSuccessfulUploadTime) > lastUpload)
 }
 
-func TestTracingReader(t *testing.T) {
-	r := bytes.NewReader([]byte("hello world"))
-	tr := newTracingReadCloser(NopCloserWithSize(r), nil)
-
-	size, err := TryToGetSize(tr)
-
-	testutil.Ok(t, err)
-	testutil.Equals(t, int64(11), size)
-
-	smallBuf := make([]byte, 4)
-	n, err := io.ReadFull(tr, smallBuf)
-	testutil.Ok(t, err)
-	testutil.Equals(t, 4, n)
-
-	// Verify that size is still the same, after reading 4 bytes.
-	size, err = TryToGetSize(tr)
-
-	testutil.Ok(t, err)
-	testutil.Equals(t, int64(11), size)
-}
-
 func TestDownloadUploadDirConcurrency(t *testing.T) {
 	r := prometheus.NewRegistry()
-	m := BucketWithMetrics("", NewInMemBucket(), r)
+	m := WrapWithMetrics(NewInMemBucket(), r, "")
 	tempDir := t.TempDir()
 
 	testutil.Ok(t, m.Upload(context.Background(), "dir/obj1", bytes.NewReader([]byte("1"))))
@@ -158,14 +137,13 @@ func TestDownloadUploadDirConcurrency(t *testing.T) {
 }
 
 func TestTimingTracingReader(t *testing.T) {
-	m := BucketWithMetrics("", NewInMemBucket(), nil)
+	m := WrapWithMetrics(NewInMemBucket(), nil, "")
 	r := bytes.NewReader([]byte("hello world"))
 
 	tr := NopCloserWithSize(r)
 	tr = newTimingReadCloser(tr, "", m.opsDuration, m.opsFailures, func(err error) bool {
 		return false
 	}, m.opsFetchedBytes)
-	tr = newTracingReadCloser(tr, nil)
 
 	size, err := TryToGetSize(tr)