swarm: integrate OpenTracing; propagate ctx to internal APIs (#17169)

* swarm: propagate ctx, enable opentracing

* swarm/tracing: log error when tracing is misconfigured

19 files changed, 179 insertions, 105 deletions

diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go
index 2d197fefa..b9b502273 100644
--- a/swarm/storage/chunker.go
+++ b/swarm/storage/chunker.go
@@ -26,6 +26,9 @@ import (
 
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/swarm/log"
+	"github.com/ethereum/go-ethereum/swarm/spancontext"
+	opentracing "github.com/opentracing/opentracing-go"
+	olog "github.com/opentracing/opentracing-go/log"
 )
 
 /*
@@ -93,9 +96,12 @@ type JoinerParams struct {
 	getter Getter
 	// TODO: there is a bug, so depth can only be 0 today, see: https://github.com/ethersphere/go-ethereum/issues/344
 	depth int
+	ctx   context.Context
 }
 
 type TreeChunker struct {
+	ctx context.Context
+
 	branches int64
 	hashFunc SwarmHasher
 	dataSize int64
@@ -136,6 +142,7 @@ func TreeJoin(ctx context.Context, addr Address, getter Getter, depth int) *Lazy
 		addr:   addr,
 		getter: getter,
 		depth:  depth,
+		ctx:    ctx,
 	}
 
 	return NewTreeJoiner(jp).Join(ctx)
@@ -174,6 +181,8 @@ func NewTreeJoiner(params *JoinerParams) *TreeChunker {
 	tc.errC = make(chan error)
 	tc.quitC = make(chan bool)
 
+	tc.ctx = params.ctx
+
 	return tc
 }
 
@@ -351,7 +360,7 @@ func (tc *TreeChunker) runWorker() {
 					return
 				}
 
-				h, err := tc.putter.Put(job.chunk)
+				h, err := tc.putter.Put(tc.ctx, job.chunk)
 				if err != nil {
 					tc.errC <- err
 					return
@@ -371,6 +380,7 @@ func (tc *TreeChunker) Append() (Address, func(), error) {
 
 // LazyChunkReader implements LazySectionReader
 type LazyChunkReader struct {
+	Ctx       context.Context
 	key       Address // root key
 	chunkData ChunkData
 	off       int64 // offset
@@ -389,16 +399,28 @@ func (tc *TreeChunker) Join(ctx context.Context) *LazyChunkReader {
 		hashSize:  tc.hashSize,
 		depth:     tc.depth,
 		getter:    tc.getter,
+		Ctx:       tc.ctx,
 	}
 }
 
+func (r *LazyChunkReader) Context() context.Context {
+	return r.Ctx
+}
+
 // Size is meant to be called on the LazySectionReader
-func (r *LazyChunkReader) Size(quitC chan bool) (n int64, err error) {
+func (r *LazyChunkReader) Size(ctx context.Context, quitC chan bool) (n int64, err error) {
 	metrics.GetOrRegisterCounter("lazychunkreader.size", nil).Inc(1)
 
+	var sp opentracing.Span
+	var cctx context.Context
+	cctx, sp = spancontext.StartSpan(
+		ctx,
+		"lcr.size")
+	defer sp.Finish()
+
 	log.Debug("lazychunkreader.size", "key", r.key)
 	if r.chunkData == nil {
-		chunkData, err := r.getter.Get(Reference(r.key))
+		chunkData, err := r.getter.Get(cctx, Reference(r.key))
 		if err != nil {
 			return 0, err
 		}
@@ -421,12 +443,25 @@ func (r *LazyChunkReader) Size(quitC chan bool) (n int64, err error) {
 func (r *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 	metrics.GetOrRegisterCounter("lazychunkreader.readat", nil).Inc(1)
 
+	var sp opentracing.Span
+	var cctx context.Context
+	cctx, sp = spancontext.StartSpan(
+		r.Ctx,
+		"lcr.read")
+	defer sp.Finish()
+
+	defer func() {
+		sp.LogFields(
+			olog.Int("off", int(off)),
+			olog.Int("read", read))
+	}()
+
 	// this is correct, a swarm doc cannot be zero length, so no EOF is expected
 	if len(b) == 0 {
 		return 0, nil
 	}
 	quitC := make(chan bool)
-	size, err := r.Size(quitC)
+	size, err := r.Size(cctx, quitC)
 	if err != nil {
 		log.Error("lazychunkreader.readat.size", "size", size, "err", err)
 		return 0, err
@@ -449,7 +484,7 @@ func (r *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 		length *= r.chunkSize
 	}
 	wg.Add(1)
-	go r.join(b, off, off+length, depth, treeSize/r.branches, r.chunkData, &wg, errC, quitC)
+	go r.join(cctx, b, off, off+length, depth, treeSize/r.branches, r.chunkData, &wg, errC, quitC)
 	go func() {
 		wg.Wait()
 		close(errC)
@@ -467,7 +502,7 @@ func (r *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
 	return len(b), nil
 }
 
-func (r *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeSize int64, chunkData ChunkData, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) {
+func (r *LazyChunkReader) join(ctx context.Context, b []byte, off int64, eoff int64, depth int, treeSize int64, chunkData ChunkData, parentWg *sync.WaitGroup, errC chan error, quitC chan bool) {
 	defer parentWg.Done()
 	// find appropriate block level
 	for chunkData.Size() < treeSize && depth > r.depth {
@@ -514,7 +549,7 @@ func (r *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeS
 		wg.Add(1)
 		go func(j int64) {
 			childKey := chunkData[8+j*r.hashSize : 8+(j+1)*r.hashSize]
-			chunkData, err := r.getter.Get(Reference(childKey))
+			chunkData, err := r.getter.Get(ctx, Reference(childKey))
 			if err != nil {
 				log.Error("lazychunkreader.join", "key", fmt.Sprintf("%x", childKey), "err", err)
 				select {
@@ -533,7 +568,7 @@ func (r *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, treeS
 			if soff < off {
 				soff = off
 			}
-			r.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/r.branches, chunkData, wg, errC, quitC)
+			r.join(ctx, b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/r.branches, chunkData, wg, errC, quitC)
 		}(i)
 	} //for
 }
@@ -570,7 +605,7 @@ func (r *LazyChunkReader) Seek(offset int64, whence int) (int64, error) {
 		offset += r.off
 	case 2:
 		if r.chunkData == nil { //seek from the end requires rootchunk for size. call Size first
-			_, err := r.Size(nil)
+			_, err := r.Size(context.TODO(), nil)
 			if err != nil {
 				return 0, fmt.Errorf("can't get size: %v", err)
 			}
diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go
index 69c388b39..dbcc8700d 100644
--- a/swarm/storage/chunker_test.go
+++ b/swarm/storage/chunker_test.go
@@ -50,11 +50,11 @@ type fakeChunkStore struct {
 }
 
 // Put doesn't store anything it is just here to implement ChunkStore
-func (f *fakeChunkStore) Put(*Chunk) {
+func (f *fakeChunkStore) Put(context.Context, *Chunk) {
 }
 
 // Gut doesn't store anything it is just here to implement ChunkStore
-func (f *fakeChunkStore) Get(Address) (*Chunk, error) {
+func (f *fakeChunkStore) Get(context.Context, Address) (*Chunk, error) {
 	return nil, errors.New("FakeChunkStore doesn't support Get")
 }
 
@@ -281,7 +281,7 @@ func TestRandomBrokenData(t *testing.T) {
 }
 
 func benchReadAll(reader LazySectionReader) {
-	size, _ := reader.Size(nil)
+	size, _ := reader.Size(context.TODO(), nil)
 	output := make([]byte, 1000)
 	for pos := int64(0); pos < size; pos += 1000 {
 		reader.ReadAt(output, pos)
diff --git a/swarm/storage/chunkstore.go b/swarm/storage/chunkstore.go
index ce95cd971..3b4d97a7a 100644
--- a/swarm/storage/chunkstore.go
+++ b/swarm/storage/chunkstore.go
@@ -16,7 +16,10 @@
 
 package storage
 
-import "sync"
+import (
+	"context"
+	"sync"
+)
 
 /*
 ChunkStore interface is implemented by :
@@ -28,8 +31,8 @@ ChunkStore interface is implemented by :
 - FakeChunkStore: dummy store which doesn't store anything just implements the interface
 */
 type ChunkStore interface {
-	Put(*Chunk) // effectively there is no error even if there is an error
-	Get(Address) (*Chunk, error)
+	Put(context.Context, *Chunk) // effectively there is no error even if there is an error
+	Get(context.Context, Address) (*Chunk, error)
 	Close()
 }
 
@@ -45,14 +48,14 @@ func NewMapChunkStore() *MapChunkStore {
 	}
 }
 
-func (m *MapChunkStore) Put(chunk *Chunk) {
+func (m *MapChunkStore) Put(ctx context.Context, chunk *Chunk) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.chunks[chunk.Addr.Hex()] = chunk
 	chunk.markAsStored()
 }
 
-func (m *MapChunkStore) Get(addr Address) (*Chunk, error) {
+func (m *MapChunkStore) Get(ctx context.Context, addr Address) (*Chunk, error) {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
 	chunk := m.chunks[addr.Hex()]
diff --git a/swarm/storage/common.go b/swarm/storage/common.go
index d86cb6914..d6352820e 100644
--- a/swarm/storage/common.go
+++ b/swarm/storage/common.go
@@ -16,6 +16,7 @@
 package storage
 
 import (
+	"context"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/swarm/log"
@@ -37,7 +38,7 @@ func PutChunks(store *LocalStore, chunks ...*Chunk) {
 		}
 	}()
 	for _, c := range chunks {
-		go store.Put(c)
+		go store.Put(context.TODO(), c)
 	}
 	wg.Wait()
 }
diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go
index c6e97d68f..dc1a3ab35 100644
--- a/swarm/storage/common_test.go
+++ b/swarm/storage/common_test.go
@@ -18,6 +18,7 @@ package storage
 
 import (
 	"bytes"
+	"context"
 	"crypto/rand"
 	"flag"
 	"fmt"
@@ -69,7 +70,7 @@ func mput(store ChunkStore, processors int, n int, f func(i int64) *Chunk) (hs [
 			for chunk := range c {
 				wg.Add(1)
 				chunk := chunk
-				store.Put(chunk)
+				store.Put(context.TODO(), chunk)
 				go func() {
 					defer wg.Done()
 					<-chunk.dbStoredC
@@ -103,7 +104,7 @@ func mget(store ChunkStore, hs []Address, f func(h Address, chunk *Chunk) error)
 	for _, k := range hs {
 		go func(h Address) {
 			defer wg.Done()
-			chunk, err := store.Get(h)
+			chunk, err := store.Get(context.TODO(), h)
 			if err != nil {
 				errc <- err
 				return
diff --git a/swarm/storage/dbapi.go b/swarm/storage/dbapi.go
index 24234b031..dd71752eb 100644
--- a/swarm/storage/dbapi.go
+++ b/swarm/storage/dbapi.go
@@ -16,6 +16,8 @@
 
 package storage
 
+import "context"
+
 // wrapper of db-s to provide mockable custom local chunk store access to syncer
 type DBAPI struct {
 	db  *LDBStore
@@ -27,8 +29,8 @@ func NewDBAPI(loc *LocalStore) *DBAPI {
 }
 
 // to obtain the chunks from address or request db entry only
-func (d *DBAPI) Get(addr Address) (*Chunk, error) {
-	return d.loc.Get(addr)
+func (d *DBAPI) Get(ctx context.Context, addr Address) (*Chunk, error) {
+	return d.loc.Get(ctx, addr)
 }
 
 // current storage counter of chunk db
@@ -42,11 +44,11 @@ func (d *DBAPI) Iterator(from uint64, to uint64, po uint8, f func(Address, uint6
 }
 
 // to obtain the chunks from address or request db entry only
-func (d *DBAPI) GetOrCreateRequest(addr Address) (*Chunk, bool) {
-	return d.loc.GetOrCreateRequest(addr)
+func (d *DBAPI) GetOrCreateRequest(ctx context.Context, addr Address) (*Chunk, bool) {
+	return d.loc.GetOrCreateRequest(ctx, addr)
 }
 
 // to obtain the chunks from key or request db entry only
-func (d *DBAPI) Put(chunk *Chunk) {
-	d.loc.Put(chunk)
+func (d *DBAPI) Put(ctx context.Context, chunk *Chunk) {
+	d.loc.Put(ctx, chunk)
 }
diff --git a/swarm/storage/hasherstore.go b/swarm/storage/hasherstore.go
index e18b66ddc..139c0ee03 100644
--- a/swarm/storage/hasherstore.go
+++ b/swarm/storage/hasherstore.go
@@ -74,7 +74,7 @@ func NewHasherStore(chunkStore ChunkStore, hashFunc SwarmHasher, toEncrypt bool)
 // Put stores the chunkData into the ChunkStore of the hasherStore and returns the reference.
 // If hasherStore has a chunkEncryption object, the data will be encrypted.
 // Asynchronous function, the data will not necessarily be stored when it returns.
-func (h *hasherStore) Put(chunkData ChunkData) (Reference, error) {
+func (h *hasherStore) Put(ctx context.Context, chunkData ChunkData) (Reference, error) {
 	c := chunkData
 	size := chunkData.Size()
 	var encryptionKey encryption.Key
@@ -87,7 +87,7 @@ func (h *hasherStore) Put(chunkData ChunkData) (Reference, error) {
 	}
 	chunk := h.createChunk(c, size)
 
-	h.storeChunk(chunk)
+	h.storeChunk(ctx, chunk)
 
 	return Reference(append(chunk.Addr, encryptionKey...)), nil
 }
@@ -95,14 +95,14 @@ func (h *hasherStore) Put(chunkData ChunkData) (Reference, error) {
 // Get returns data of the chunk with the given reference (retrieved from the ChunkStore of hasherStore).
 // If the data is encrypted and the reference contains an encryption key, it will be decrypted before
 // return.
-func (h *hasherStore) Get(ref Reference) (ChunkData, error) {
+func (h *hasherStore) Get(ctx context.Context, ref Reference) (ChunkData, error) {
 	key, encryptionKey, err := parseReference(ref, h.hashSize)
 	if err != nil {
 		return nil, err
 	}
 	toDecrypt := (encryptionKey != nil)
 
-	chunk, err := h.store.Get(key)
+	chunk, err := h.store.Get(ctx, key)
 	if err != nil {
 		return nil, err
 	}
@@ -207,13 +207,13 @@ func (h *hasherStore) RefSize() int64 {
 	return h.refSize
 }
 
-func (h *hasherStore) storeChunk(chunk *Chunk) {
+func (h *hasherStore) storeChunk(ctx context.Context, chunk *Chunk) {
 	h.wg.Add(1)
 	go func() {
 		<-chunk.dbStoredC
 		h.wg.Done()
 	}()
-	h.store.Put(chunk)
+	h.store.Put(ctx, chunk)
 }
 
 func parseReference(ref Reference, hashSize int) (Address, encryption.Key, error) {
diff --git a/swarm/storage/hasherstore_test.go b/swarm/storage/hasherstore_test.go
index cf7b0dcc3..ddf1c39b0 100644
--- a/swarm/storage/hasherstore_test.go
+++ b/swarm/storage/hasherstore_test.go
@@ -47,13 +47,13 @@ func TestHasherStore(t *testing.T) {
 
 		// Put two random chunks into the hasherStore
 		chunkData1 := GenerateRandomChunk(int64(tt.chunkLength)).SData
-		key1, err := hasherStore.Put(chunkData1)
+		key1, err := hasherStore.Put(context.TODO(), chunkData1)
 		if err != nil {
 			t.Fatalf("Expected no error got \"%v\"", err)
 		}
 
 		chunkData2 := GenerateRandomChunk(int64(tt.chunkLength)).SData
-		key2, err := hasherStore.Put(chunkData2)
+		key2, err := hasherStore.Put(context.TODO(), chunkData2)
 		if err != nil {
 			t.Fatalf("Expected no error got \"%v\"", err)
 		}
@@ -67,7 +67,7 @@ func TestHasherStore(t *testing.T) {
 		}
 
 		// Get the first chunk
-		retrievedChunkData1, err := hasherStore.Get(key1)
+		retrievedChunkData1, err := hasherStore.Get(context.TODO(), key1)
 		if err != nil {
 			t.Fatalf("Expected no error, got \"%v\"", err)
 		}
@@ -78,7 +78,7 @@ func TestHasherStore(t *testing.T) {
 		}
 
 		// Get the second chunk
-		retrievedChunkData2, err := hasherStore.Get(key2)
+		retrievedChunkData2, err := hasherStore.Get(context.TODO(), key2)
 		if err != nil {
 			t.Fatalf("Expected no error, got \"%v\"", err)
 		}
@@ -105,7 +105,7 @@ func TestHasherStore(t *testing.T) {
 		}
 
 		// Check if chunk data in store is encrypted or not
-		chunkInStore, err := chunkStore.Get(hash1)
+		chunkInStore, err := chunkStore.Get(context.TODO(), hash1)
 		if err != nil {
 			t.Fatalf("Expected no error got \"%v\"", err)
 		}
diff --git a/swarm/storage/ldbstore.go b/swarm/storage/ldbstore.go
index 178b1ebc4..7920ee767 100644
--- a/swarm/storage/ldbstore.go
+++ b/swarm/storage/ldbstore.go
@@ -25,6 +25,7 @@ package storage
 import (
 	"archive/tar"
 	"bytes"
+	"context"
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
@@ -370,7 +371,7 @@ func (s *LDBStore) Import(in io.Reader) (int64, error) {
 		key := Address(keybytes)
 		chunk := NewChunk(key, nil)
 		chunk.SData = data[32:]
-		s.Put(chunk)
+		s.Put(context.TODO(), chunk)
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
@@ -499,7 +500,7 @@ func (s *LDBStore) CurrentStorageIndex() uint64 {
 	return s.dataIdx
 }
 
-func (s *LDBStore) Put(chunk *Chunk) {
+func (s *LDBStore) Put(ctx context.Context, chunk *Chunk) {
 	metrics.GetOrRegisterCounter("ldbstore.put", nil).Inc(1)
 	log.Trace("ldbstore.put", "key", chunk.Addr)
 
@@ -639,7 +640,7 @@ func (s *LDBStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
 	return true
 }
 
-func (s *LDBStore) Get(addr Address) (chunk *Chunk, err error) {
+func (s *LDBStore) Get(ctx context.Context, addr Address) (chunk *Chunk, err error) {
 	metrics.GetOrRegisterCounter("ldbstore.get", nil).Inc(1)
 	log.Trace("ldbstore.get", "key", addr)
 
diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go
index 2453d2f30..baf9e8c14 100644
--- a/swarm/storage/ldbstore_test.go
+++ b/swarm/storage/ldbstore_test.go
@@ -18,6 +18,7 @@ package storage
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -157,7 +158,7 @@ func testDbStoreNotFound(t *testing.T, mock bool) {
 		t.Fatalf("init dbStore failed: %v", err)
 	}
 
-	_, err = db.Get(ZeroAddr)
+	_, err = db.Get(context.TODO(), ZeroAddr)
 	if err != ErrChunkNotFound {
 		t.Errorf("Expected ErrChunkNotFound, got %v", err)
 	}
@@ -188,7 +189,7 @@ func testIterator(t *testing.T, mock bool) {
 	wg := &sync.WaitGroup{}
 	wg.Add(len(chunks))
 	for i = 0; i < len(chunks); i++ {
-		db.Put(chunks[i])
+		db.Put(context.TODO(), chunks[i])
 		chunkkeys[i] = chunks[i].Addr
 		j := i
 		go func() {
@@ -299,7 +300,7 @@ func TestLDBStoreWithoutCollectGarbage(t *testing.T) {
 	}
 
 	for i := 0; i < n; i++ {
-		go ldb.Put(chunks[i])
+		go ldb.Put(context.TODO(), chunks[i])
 	}
 
 	// wait for all chunks to be stored
@@ -310,7 +311,7 @@ func TestLDBStoreWithoutCollectGarbage(t *testing.T) {
 	log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt)
 
 	for i := 0; i < n; i++ {
-		ret, err := ldb.Get(chunks[i].Addr)
+		ret, err := ldb.Get(context.TODO(), chunks[i].Addr)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -349,7 +350,7 @@ func TestLDBStoreCollectGarbage(t *testing.T) {
 	}
 
 	for i := 0; i < n; i++ {
-		ldb.Put(chunks[i])
+		ldb.Put(context.TODO(), chunks[i])
 	}
 
 	// wait for all chunks to be stored
@@ -364,7 +365,7 @@ func TestLDBStoreCollectGarbage(t *testing.T) {
 
 	var missing int
 	for i := 0; i < n; i++ {
-		ret, err := ldb.Get(chunks[i].Addr)
+		ret, err := ldb.Get(context.TODO(), chunks[i].Addr)
 		if err == ErrChunkNotFound || err == ldberrors.ErrNotFound {
 			missing++
 			continue
@@ -403,7 +404,7 @@ func TestLDBStoreAddRemove(t *testing.T) {
 	}
 
 	for i := 0; i < n; i++ {
-		go ldb.Put(chunks[i])
+		go ldb.Put(context.TODO(), chunks[i])
 	}
 
 	// wait for all chunks to be stored before continuing
@@ -428,7 +429,7 @@ func TestLDBStoreAddRemove(t *testing.T) {
 	log.Info("ldbstore", "entrycnt", ldb.entryCnt, "accesscnt", ldb.accessCnt)
 
 	for i := 0; i < n; i++ {
-		ret, err := ldb.Get(chunks[i].Addr)
+		ret, err := ldb.Get(context.TODO(), chunks[i].Addr)
 
 		if i%2 == 0 {
 			// expect even chunks to be missing
@@ -465,7 +466,7 @@ func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) {
 	}
 
 	for i := 0; i < n; i++ {
-		ldb.Put(chunks[i])
+		ldb.Put(context.TODO(), chunks[i])
 	}
 
 	// wait for all chunks to be stored before continuing
@@ -494,7 +495,7 @@ func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) {
 	n = 10
 
 	for i := 0; i < n; i++ {
-		ldb.Put(chunks[i])
+		ldb.Put(context.TODO(), chunks[i])
 	}
 
 	// wait for all chunks to be stored before continuing
@@ -504,14 +505,14 @@ func TestLDBStoreRemoveThenCollectGarbage(t *testing.T) {
 
 	// expect for first chunk to be missing, because it has the smallest access value
 	idx := 0
-	ret, err := ldb.Get(chunks[idx].Addr)
+	ret, err := ldb.Get(context.TODO(), chunks[idx].Addr)
 	if err == nil || ret != nil {
 		t.Fatal("expected first chunk to be missing, but got no error")
 	}
 
 	// expect for last chunk to be present, as it has the largest access value
 	idx = 9
-	ret, err = ldb.Get(chunks[idx].Addr)
+	ret, err = ldb.Get(context.TODO(), chunks[idx].Addr)
 	if err != nil {
 		t.Fatalf("expected no error, but got %s", err)
 	}
diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go
index 4c57086fa..096d150ae 100644
--- a/swarm/storage/localstore.go
+++ b/swarm/storage/localstore.go
@@ -17,6 +17,7 @@
 package storage
 
 import (
+	"context"
 	"encoding/binary"
 	"fmt"
 	"path/filepath"
@@ -96,7 +97,7 @@ func NewTestLocalStoreForAddr(params *LocalStoreParams) (*LocalStore, error) {
 // when the chunk is stored in memstore.
 // After the LDBStore.Put, it is ensured that the MemStore
 // contains the chunk with the same data, but nil ReqC channel.
-func (ls *LocalStore) Put(chunk *Chunk) {
+func (ls *LocalStore) Put(ctx context.Context, chunk *Chunk) {
 	if l := len(chunk.SData); l < 9 {
 		log.Debug("incomplete chunk data", "addr", chunk.Addr, "length", l)
 		chunk.SetErrored(ErrChunkInvalid)
@@ -123,7 +124,7 @@ func (ls *LocalStore) Put(chunk *Chunk) {
 
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
 
-	memChunk, err := ls.memStore.Get(chunk.Addr)
+	memChunk, err := ls.memStore.Get(ctx, chunk.Addr)
 	switch err {
 	case nil:
 		if memChunk.ReqC == nil {
@@ -136,7 +137,7 @@ func (ls *LocalStore) Put(chunk *Chunk) {
 		return
 	}
 
-	ls.DbStore.Put(chunk)
+	ls.DbStore.Put(ctx, chunk)
 
 	// chunk is no longer a request, but a chunk with data, so replace it in memStore
 	newc := NewChunk(chunk.Addr, nil)
@@ -144,7 +145,7 @@ func (ls *LocalStore) Put(chunk *Chunk) {
 	newc.Size = chunk.Size
 	newc.dbStoredC = chunk.dbStoredC
 
-	ls.memStore.Put(newc)
+	ls.memStore.Put(ctx, newc)
 
 	if memChunk != nil && memChunk.ReqC != nil {
 		close(memChunk.ReqC)
@@ -155,15 +156,15 @@ func (ls *LocalStore) Put(chunk *Chunk) {
 // This method is blocking until the chunk is retrieved
 // so additional timeout may be needed to wrap this call if
 // ChunkStores are remote and can have long latency
-func (ls *LocalStore) Get(addr Address) (chunk *Chunk, err error) {
+func (ls *LocalStore) Get(ctx context.Context, addr Address) (chunk *Chunk, err error) {
 	ls.mu.Lock()
 	defer ls.mu.Unlock()
 
-	return ls.get(addr)
+	return ls.get(ctx, addr)
 }
 
-func (ls *LocalStore) get(addr Address) (chunk *Chunk, err error) {
-	chunk, err = ls.memStore.Get(addr)
+func (ls *LocalStore) get(ctx context.Context, addr Address) (chunk *Chunk, err error) {
+	chunk, err = ls.memStore.Get(ctx, addr)
 	if err == nil {
 		if chunk.ReqC != nil {
 			select {
@@ -177,25 +178,25 @@ func (ls *LocalStore) get(addr Address) (chunk *Chunk, err error) {
 		return
 	}
 	metrics.GetOrRegisterCounter("localstore.get.cachemiss", nil).Inc(1)
-	chunk, err = ls.DbStore.Get(addr)
+	chunk, err = ls.DbStore.Get(ctx, addr)
 	if err != nil {
 		metrics.GetOrRegisterCounter("localstore.get.error", nil).Inc(1)
 		return
 	}
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
-	ls.memStore.Put(chunk)
+	ls.memStore.Put(ctx, chunk)
 	return
 }
 
 // retrieve logic common for local and network chunk retrieval requests
-func (ls *LocalStore) GetOrCreateRequest(addr Address) (chunk *Chunk, created bool) {
+func (ls *LocalStore) GetOrCreateRequest(ctx context.Context, addr Address) (chunk *Chunk, created bool) {
 	metrics.GetOrRegisterCounter("localstore.getorcreaterequest", nil).Inc(1)
 
 	ls.mu.Lock()
 	defer ls.mu.Unlock()
 
 	var err error
-	chunk, err = ls.get(addr)
+	chunk, err = ls.get(ctx, addr)
 	if err == nil && chunk.GetErrored() == nil {
 		metrics.GetOrRegisterCounter("localstore.getorcreaterequest.hit", nil).Inc(1)
 		log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v found locally", addr))
@@ -210,7 +211,7 @@ func (ls *LocalStore) GetOrCreateRequest(addr Address) (chunk *Chunk, created bo
 	metrics.GetOrRegisterCounter("localstore.getorcreaterequest.miss", nil).Inc(1)
 	log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v not found locally. open new request", addr))
 	chunk = NewChunk(addr, make(chan bool))
-	ls.memStore.Put(chunk)
+	ls.memStore.Put(ctx, chunk)
 	return chunk, true
 }
 
diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go
index 7af31ffbd..55cfcbfea 100644
--- a/swarm/storage/memstore.go
+++ b/swarm/storage/memstore.go
@@ -19,6 +19,7 @@
 package storage
 
 import (
+	"context"
 	"sync"
 
 	lru "github.com/hashicorp/golang-lru"
@@ -68,7 +69,7 @@ func NewMemStore(params *StoreParams, _ *LDBStore) (m *MemStore) {
 	}
 }
 
-func (m *MemStore) Get(addr Address) (*Chunk, error) {
+func (m *MemStore) Get(ctx context.Context, addr Address) (*Chunk, error) {
 	if m.disabled {
 		return nil, ErrChunkNotFound
 	}
@@ -90,7 +91,7 @@ func (m *MemStore) Get(addr Address) (*Chunk, error) {
 	return c.(*Chunk), nil
 }
 
-func (m *MemStore) Put(c *Chunk) {
+func (m *MemStore) Put(ctx context.Context, c *Chunk) {
 	if m.disabled {
 		return
 	}
diff --git a/swarm/storage/memstore_test.go b/swarm/storage/memstore_test.go
index 5c68a4b4b..2c1b0e89e 100644
--- a/swarm/storage/memstore_test.go
+++ b/swarm/storage/memstore_test.go
@@ -17,6 +17,7 @@
 package storage
 
 import (
+	"context"
 	"crypto/rand"
 	"encoding/binary"
 	"io/ioutil"
@@ -72,7 +73,7 @@ func TestMemStoreNotFound(t *testing.T) {
 	m := newTestMemStore()
 	defer m.Close()
 
-	_, err := m.Get(ZeroAddr)
+	_, err := m.Get(context.TODO(), ZeroAddr)
 	if err != ErrChunkNotFound {
 		t.Errorf("Expected ErrChunkNotFound, got %v", err)
 	}
@@ -187,8 +188,8 @@ func TestMemStoreAndLDBStore(t *testing.T) {
 		}
 
 		for i := 0; i < tt.n; i++ {
-			go ldb.Put(chunks[i])
-			memStore.Put(chunks[i])
+			go ldb.Put(context.TODO(), chunks[i])
+			memStore.Put(context.TODO(), chunks[i])
 
 			if got := memStore.cache.Len(); got > cacheCap {
 				t.Fatalf("expected to get cache capacity less than %v, but got %v", cacheCap, got)
@@ -200,10 +201,10 @@ func TestMemStoreAndLDBStore(t *testing.T) {
 		}
 
 		for i := 0; i < tt.n; i++ {
-			_, err := memStore.Get(chunks[i].Addr)
+			_, err := memStore.Get(context.TODO(), chunks[i].Addr)
 			if err != nil {
 				if err == ErrChunkNotFound {
-					_, err := ldb.Get(chunks[i].Addr)
+					_, err := ldb.Get(context.TODO(), chunks[i].Addr)
 					if err != nil {
 						t.Fatalf("couldn't get chunk %v from ldb, got error: %v", i, err)
 					}
diff --git a/swarm/storage/mru/resource.go b/swarm/storage/mru/resource.go
index 1e92a5e92..4f5a4f44c 100644
--- a/swarm/storage/mru/resource.go
+++ b/swarm/storage/mru/resource.go
@@ -125,6 +125,10 @@ type resource struct {
 	updated    time.Time
 }
 
+func (r *resource) Context() context.Context {
+	return context.TODO()
+}
+
 // TODO Expire content after a defined period (to force resync)
 func (r *resource) isSynced() bool {
 	return !r.updated.IsZero()
@@ -134,7 +138,7 @@ func (r *resource) NameHash() common.Hash {
 	return r.nameHash
 }
 
-func (r *resource) Size(chan bool) (int64, error) {
+func (r *resource) Size(context.Context, chan bool) (int64, error) {
 	if !r.isSynced() {
 		return 0, NewError(ErrNotSynced, "Not synced")
 	}
@@ -413,7 +417,7 @@ func (h *Handler) New(ctx context.Context, name string, frequency uint64) (stora
 
 	chunk := h.newMetaChunk(name, currentblock, frequency)
 
-	h.chunkStore.Put(chunk)
+	h.chunkStore.Put(ctx, chunk)
 	log.Debug("new resource", "name", name, "key", nameHash, "startBlock", currentblock, "frequency", frequency)
 
 	// create the internal index for the resource and populate it with the data of the first version
@@ -593,7 +597,7 @@ func (h *Handler) lookup(rsrc *resource, period uint32, version uint32, refresh
 			return nil, NewError(ErrPeriodDepth, fmt.Sprintf("Lookup exceeded max period hops (%d)", maxLookup.Max))
 		}
 		key := h.resourceHash(period, version, rsrc.nameHash)
-		chunk, err := h.chunkStore.GetWithTimeout(key, defaultRetrieveTimeout)
+		chunk, err := h.chunkStore.GetWithTimeout(context.TODO(), key, defaultRetrieveTimeout)
 		if err == nil {
 			if specificversion {
 				return h.updateIndex(rsrc, chunk)
@@ -603,7 +607,7 @@ func (h *Handler) lookup(rsrc *resource, period uint32, version uint32, refresh
 			for {
 				newversion := version + 1
 				key := h.resourceHash(period, newversion, rsrc.nameHash)
-				newchunk, err := h.chunkStore.GetWithTimeout(key, defaultRetrieveTimeout)
+				newchunk, err := h.chunkStore.GetWithTimeout(context.TODO(), key, defaultRetrieveTimeout)
 				if err != nil {
 					return h.updateIndex(rsrc, chunk)
 				}
@@ -621,8 +625,8 @@ func (h *Handler) lookup(rsrc *resource, period uint32, version uint32, refresh
 
 // Retrieves a resource metadata chunk and creates/updates the index entry for it
 // with the resulting metadata
-func (h *Handler) Load(addr storage.Address) (*resource, error) {
-	chunk, err := h.chunkStore.GetWithTimeout(addr, defaultRetrieveTimeout)
+func (h *Handler) Load(ctx context.Context, addr storage.Address) (*resource, error) {
+	chunk, err := h.chunkStore.GetWithTimeout(ctx, addr, defaultRetrieveTimeout)
 	if err != nil {
 		return nil, NewError(ErrNotFound, err.Error())
 	}
@@ -890,7 +894,7 @@ func (h *Handler) update(ctx context.Context, name string, data []byte, multihas
 	chunk := newUpdateChunk(key, signature, nextperiod, version, name, data, datalength)
 
 	// send the chunk
-	h.chunkStore.Put(chunk)
+	h.chunkStore.Put(ctx, chunk)
 	log.Trace("resource update", "name", name, "key", key, "currentblock", currentblock, "lastperiod", nextperiod, "version", version, "data", chunk.SData, "multihash", multihash)
 
 	// update our resources map entry and return the new key
diff --git a/swarm/storage/mru/resource_test.go b/swarm/storage/mru/resource_test.go
index aa1860359..48387d981 100644
--- a/swarm/storage/mru/resource_test.go
+++ b/swarm/storage/mru/resource_test.go
@@ -182,7 +182,7 @@ func TestHandler(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	chunk, err := rh.chunkStore.Get(storage.Address(rootChunkKey))
+	chunk, err := rh.chunkStore.Get(context.TODO(), storage.Address(rootChunkKey))
 	if err != nil {
 		t.Fatal(err)
 	} else if len(chunk.SData) < 16 {
@@ -256,7 +256,7 @@ func TestHandler(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	rsrc2, err := rh2.Load(rootChunkKey)
+	rsrc2, err := rh2.Load(context.TODO(), rootChunkKey)
 	_, err = rh2.LookupLatest(ctx, nameHash, true, nil)
 	if err != nil {
 		t.Fatal(err)
@@ -754,7 +754,7 @@ func newTestSigner() (*GenericSigner, error) {
 }
 
 func getUpdateDirect(rh *Handler, addr storage.Address) ([]byte, error) {
-	chunk, err := rh.chunkStore.Get(addr)
+	chunk, err := rh.chunkStore.Get(context.TODO(), addr)
 	if err != nil {
 		return nil, err
 	}
diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go
index 6a205cfa4..96a7e51f7 100644
--- a/swarm/storage/netstore.go
+++ b/swarm/storage/netstore.go
@@ -17,9 +17,12 @@
 package storage
 
 import (
+	"context"
 	"time"
 
 	"github.com/ethereum/go-ethereum/swarm/log"
+	"github.com/ethereum/go-ethereum/swarm/spancontext"
+	opentracing "github.com/opentracing/opentracing-go"
 )
 
 var (
@@ -43,10 +46,10 @@ var (
 // access by calling network is blocking with a timeout
 type NetStore struct {
 	localStore *LocalStore
-	retrieve   func(chunk *Chunk) error
+	retrieve   func(ctx context.Context, chunk *Chunk) error
 }
 
-func NewNetStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *NetStore {
+func NewNetStore(localStore *LocalStore, retrieve func(ctx context.Context, chunk *Chunk) error) *NetStore {
 	return &NetStore{localStore, retrieve}
 }
 
@@ -56,7 +59,14 @@ func NewNetStore(localStore *LocalStore, retrieve func(chunk *Chunk) error) *Net
 // Get uses get method to retrieve request, but retries if the
 // ErrChunkNotFound is returned by get, until the netStoreRetryTimeout
 // is reached.
-func (ns *NetStore) Get(addr Address) (chunk *Chunk, err error) {
+func (ns *NetStore) Get(ctx context.Context, addr Address) (chunk *Chunk, err error) {
+
+	var sp opentracing.Span
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"netstore.get.global")
+	defer sp.Finish()
+
 	timer := time.NewTimer(netStoreRetryTimeout)
 	defer timer.Stop()
 
@@ -84,7 +94,7 @@ func (ns *NetStore) Get(addr Address) (chunk *Chunk, err error) {
 		defer limiter.Stop()
 
 		for {
-			chunk, err := ns.get(addr, 0)
+			chunk, err := ns.get(ctx, addr, 0)
 			if err != ErrChunkNotFound {
 				// break retry only if the error is nil
 				// or other error then ErrChunkNotFound
@@ -122,16 +132,23 @@ func (ns *NetStore) Get(addr Address) (chunk *Chunk, err error) {
 }
 
 // GetWithTimeout makes a single retrieval attempt for a chunk with a explicit timeout parameter
-func (ns *NetStore) GetWithTimeout(addr Address, timeout time.Duration) (chunk *Chunk, err error) {
-	return ns.get(addr, timeout)
+func (ns *NetStore) GetWithTimeout(ctx context.Context, addr Address, timeout time.Duration) (chunk *Chunk, err error) {
+	return ns.get(ctx, addr, timeout)
 }
 
-func (ns *NetStore) get(addr Address, timeout time.Duration) (chunk *Chunk, err error) {
+func (ns *NetStore) get(ctx context.Context, addr Address, timeout time.Duration) (chunk *Chunk, err error) {
 	if timeout == 0 {
 		timeout = searchTimeout
 	}
+
+	var sp opentracing.Span
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"netstore.get")
+	defer sp.Finish()
+
 	if ns.retrieve == nil {
-		chunk, err = ns.localStore.Get(addr)
+		chunk, err = ns.localStore.Get(ctx, addr)
 		if err == nil {
 			return chunk, nil
 		}
@@ -140,14 +157,14 @@ func (ns *NetStore) get(addr Address, timeout time.Duration) (chunk *Chunk, err
 		}
 	} else {
 		var created bool
-		chunk, created = ns.localStore.GetOrCreateRequest(addr)
+		chunk, created = ns.localStore.GetOrCreateRequest(ctx, addr)
 
 		if chunk.ReqC == nil {
 			return chunk, nil
 		}
 
 		if created {
-			err := ns.retrieve(chunk)
+			err := ns.retrieve(ctx, chunk)
 			if err != nil {
 				// mark chunk request as failed so that we can retry it later
 				chunk.SetErrored(ErrChunkUnavailable)
@@ -171,8 +188,8 @@ func (ns *NetStore) get(addr Address, timeout time.Duration) (chunk *Chunk, err
 }
 
 // Put is the entrypoint for local store requests coming from storeLoop
-func (ns *NetStore) Put(chunk *Chunk) {
-	ns.localStore.Put(chunk)
+func (ns *NetStore) Put(ctx context.Context, chunk *Chunk) {
+	ns.localStore.Put(ctx, chunk)
 }
 
 // Close chunk store
diff --git a/swarm/storage/netstore_test.go b/swarm/storage/netstore_test.go
index 432a799d8..7babbf5e0 100644
--- a/swarm/storage/netstore_test.go
+++ b/swarm/storage/netstore_test.go
@@ -17,6 +17,7 @@
 package storage
 
 import (
+	"context"
 	"encoding/hex"
 	"errors"
 	"io/ioutil"
@@ -46,7 +47,7 @@ func newDummyChunk(addr Address) *Chunk {
 	return chunk
 }
 
-func (m *mockRetrieve) retrieve(chunk *Chunk) error {
+func (m *mockRetrieve) retrieve(ctx context.Context, chunk *Chunk) error {
 	hkey := hex.EncodeToString(chunk.Addr)
 	m.requests[hkey] += 1
 
@@ -100,7 +101,7 @@ func TestNetstoreFailedRequest(t *testing.T) {
 	// }
 
 	// second call
-	_, err = netStore.Get(key)
+	_, err = netStore.Get(context.TODO(), key)
 	if got := r.requests[hex.EncodeToString(key)]; got != 2 {
 		t.Fatalf("expected to have called retrieve two times, but got: %v", got)
 	}
@@ -109,7 +110,7 @@ func TestNetstoreFailedRequest(t *testing.T) {
 	}
 
 	// third call
-	chunk, err := netStore.Get(key)
+	chunk, err := netStore.Get(context.TODO(), key)
 	if got := r.requests[hex.EncodeToString(key)]; got != 3 {
 		t.Fatalf("expected to have called retrieve three times, but got: %v", got)
 	}
diff --git a/swarm/storage/pyramid.go b/swarm/storage/pyramid.go
index 6643e989a..2923c81c5 100644
--- a/swarm/storage/pyramid.go
+++ b/swarm/storage/pyramid.go
@@ -287,7 +287,7 @@ func (pc *PyramidChunker) processor(id int64) {
 func (pc *PyramidChunker) processChunk(id int64, job *chunkJob) {
 	log.Debug("pyramid.chunker: processChunk()", "id", id)
 
-	ref, err := pc.putter.Put(job.chunk)
+	ref, err := pc.putter.Put(context.TODO(), job.chunk)
 	if err != nil {
 		pc.errC <- err
 	}
@@ -302,7 +302,7 @@ func (pc *PyramidChunker) processChunk(id int64, job *chunkJob) {
 func (pc *PyramidChunker) loadTree() error {
 	log.Debug("pyramid.chunker: loadTree()")
 	// Get the root chunk to get the total size
-	chunkData, err := pc.getter.Get(Reference(pc.key))
+	chunkData, err := pc.getter.Get(context.TODO(), Reference(pc.key))
 	if err != nil {
 		return errLoadingTreeRootChunk
 	}
@@ -355,7 +355,7 @@ func (pc *PyramidChunker) loadTree() error {
 			branchCount = int64(len(ent.chunk)-8) / pc.hashSize
 			for i := int64(0); i < branchCount; i++ {
 				key := ent.chunk[8+(i*pc.hashSize) : 8+((i+1)*pc.hashSize)]
-				newChunkData, err := pc.getter.Get(Reference(key))
+				newChunkData, err := pc.getter.Get(context.TODO(), Reference(key))
 				if err != nil {
 					return errLoadingTreeChunk
 				}
@@ -417,7 +417,7 @@ func (pc *PyramidChunker) prepareChunks(isAppend bool) {
 			lastKey := parent.chunk[8+lastBranch*pc.hashSize : 8+(lastBranch+1)*pc.hashSize]
 
 			var err error
-			unfinishedChunkData, err = pc.getter.Get(lastKey)
+			unfinishedChunkData, err = pc.getter.Get(context.TODO(), lastKey)
 			if err != nil {
 				pc.errC <- err
 			}
diff --git a/swarm/storage/types.go b/swarm/storage/types.go
index 32880ead7..3114ef576 100644
--- a/swarm/storage/types.go
+++ b/swarm/storage/types.go
@@ -250,7 +250,8 @@ func GenerateRandomChunks(dataSize int64, count int) (chunks []*Chunk) {
 
 // Size, Seek, Read, ReadAt
 type LazySectionReader interface {
-	Size(chan bool) (int64, error)
+	Context() context.Context
+	Size(context.Context, chan bool) (int64, error)
 	io.Seeker
 	io.Reader
 	io.ReaderAt
@@ -260,10 +261,14 @@ type LazyTestSectionReader struct {
 	*io.SectionReader
 }
 
-func (r *LazyTestSectionReader) Size(chan bool) (int64, error) {
+func (r *LazyTestSectionReader) Size(context.Context, chan bool) (int64, error) {
 	return r.SectionReader.Size(), nil
 }
 
+func (r *LazyTestSectionReader) Context() context.Context {
+	return context.TODO()
+}
+
 type StoreParams struct {
 	Hash                       SwarmHasher `toml:"-"`
 	DbCapacity                 uint64
@@ -298,7 +303,7 @@ type Reference []byte
 
 // Putter is responsible to store data and create a reference for it
 type Putter interface {
-	Put(ChunkData) (Reference, error)
+	Put(context.Context, ChunkData) (Reference, error)
 	// RefSize returns the length of the Reference created by this Putter
 	RefSize() int64
 	// Close is to indicate that no more chunk data will be Put on this Putter
@@ -309,7 +314,7 @@ type Putter interface {
 
 // Getter is an interface to retrieve a chunk's data by its reference
 type Getter interface {
-	Get(Reference) (ChunkData, error)
+	Get(context.Context, Reference) (ChunkData, error)
 }
 
 // NOTE: this returns invalid data if chunk is encrypted