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

* swarm: propagate ctx, enable opentracing

* swarm/tracing: log error when tracing is misconfigured

6 files changed, 107 insertions, 32 deletions

diff --git a/swarm/api/api.go b/swarm/api/api.go
index ff29877ff..74af669c9 100644
--- a/swarm/api/api.go
+++ b/swarm/api/api.go
@@ -37,8 +37,10 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/swarm/log"
 	"github.com/ethereum/go-ethereum/swarm/multihash"
+	"github.com/ethereum/go-ethereum/swarm/spancontext"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/ethereum/go-ethereum/swarm/storage/mru"
+	opentracing "github.com/opentracing/opentracing-go"
 )
 
 var (
@@ -263,6 +265,12 @@ func (a *API) Resolve(ctx context.Context, uri *URI) (storage.Address, error) {
 	apiResolveCount.Inc(1)
 	log.Trace("resolving", "uri", uri.Addr)
 
+	var sp opentracing.Span
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"api.resolve")
+	defer sp.Finish()
+
 	// if the URI is immutable, check if the address looks like a hash
 	if uri.Immutable() {
 		key := uri.Address()
@@ -347,7 +355,7 @@ func (a *API) Get(ctx context.Context, manifestAddr storage.Address, path string
 			log.Trace("resource type", "key", manifestAddr, "hash", entry.Hash)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
-			rsrc, err := a.resource.Load(storage.Address(common.FromHex(entry.Hash)))
+			rsrc, err := a.resource.Load(ctx, storage.Address(common.FromHex(entry.Hash)))
 			if err != nil {
 				apiGetNotFound.Inc(1)
 				status = http.StatusNotFound
@@ -486,7 +494,7 @@ func (a *API) GetDirectoryTar(ctx context.Context, uri *URI) (io.ReadCloser, err
 
 			// retrieve the entry's key and size
 			reader, _ := a.Retrieve(ctx, storage.Address(common.Hex2Bytes(entry.Hash)))
-			size, err := reader.Size(nil)
+			size, err := reader.Size(ctx, nil)
 			if err != nil {
 				return err
 			}
@@ -883,7 +891,7 @@ func (a *API) BuildDirectoryTree(ctx context.Context, mhash string, nameresolver
 // ResourceLookup Looks up mutable resource updates at specific periods and versions
 func (a *API) ResourceLookup(ctx context.Context, addr storage.Address, period uint32, version uint32, maxLookup *mru.LookupParams) (string, []byte, error) {
 	var err error
-	rsrc, err := a.resource.Load(addr)
+	rsrc, err := a.resource.Load(ctx, addr)
 	if err != nil {
 		return "", nil, err
 	}
diff --git a/swarm/api/api_test.go b/swarm/api/api_test.go
index d1fd49b5b..78fab9508 100644
--- a/swarm/api/api_test.go
+++ b/swarm/api/api_test.go
@@ -90,7 +90,7 @@ func testGet(t *testing.T, api *API, bzzhash, path string) *testResponse {
 		t.Fatalf("unexpected error: %v", err)
 	}
 	quitC := make(chan bool)
-	size, err := reader.Size(quitC)
+	size, err := reader.Size(context.TODO(), quitC)
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go
index adf6bfbaf..aacd26699 100644
--- a/swarm/api/filesystem.go
+++ b/swarm/api/filesystem.go
@@ -277,7 +277,7 @@ func retrieveToFile(quitC chan bool, fileStore *storage.FileStore, addr storage.
 	}
 	reader, _ := fileStore.Retrieve(context.TODO(), addr)
 	writer := bufio.NewWriter(f)
-	size, err := reader.Size(quitC)
+	size, err := reader.Size(context.TODO(), quitC)
 	if err != nil {
 		return err
 	}
diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go
index 22c091026..5a7c9e93e 100644
--- a/swarm/api/http/server.go
+++ b/swarm/api/http/server.go
@@ -42,8 +42,11 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/swarm/api"
 	"github.com/ethereum/go-ethereum/swarm/log"
+	"github.com/ethereum/go-ethereum/swarm/spancontext"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"github.com/ethereum/go-ethereum/swarm/storage/mru"
+	opentracing "github.com/opentracing/opentracing-go"
+
 	"github.com/pborman/uuid"
 	"github.com/rs/cors"
 )
@@ -263,6 +266,13 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
 
 	postRawCount.Inc(1)
 
+	ctx := r.Context()
+	var sp opentracing.Span
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"http.post.raw")
+	defer sp.Finish()
+
 	toEncrypt := false
 	if r.uri.Addr == "encrypt" {
 		toEncrypt = true
@@ -286,7 +296,7 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
 		return
 	}
 
-	addr, _, err := s.api.Store(r.Context(), r.Body, r.ContentLength, toEncrypt)
+	addr, _, err := s.api.Store(ctx, r.Body, r.ContentLength, toEncrypt)
 	if err != nil {
 		postRawFail.Inc(1)
 		Respond(w, r, err.Error(), http.StatusInternalServerError)
@@ -307,8 +317,15 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
 // resulting manifest hash as a text/plain response
 func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 	log.Debug("handle.post.files", "ruid", r.ruid)
-
 	postFilesCount.Inc(1)
+
+	var sp opentracing.Span
+	ctx := r.Context()
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"http.post.files")
+	defer sp.Finish()
+
 	contentType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
 	if err != nil {
 		postFilesFail.Inc(1)
@@ -323,7 +340,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 
 	var addr storage.Address
 	if r.uri.Addr != "" && r.uri.Addr != "encrypt" {
-		addr, err = s.api.Resolve(r.Context(), r.uri)
+		addr, err = s.api.Resolve(ctx, r.uri)
 		if err != nil {
 			postFilesFail.Inc(1)
 			Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusInternalServerError)
@@ -331,7 +348,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 		}
 		log.Debug("resolved key", "ruid", r.ruid, "key", addr)
 	} else {
-		addr, err = s.api.NewManifest(r.Context(), toEncrypt)
+		addr, err = s.api.NewManifest(ctx, toEncrypt)
 		if err != nil {
 			postFilesFail.Inc(1)
 			Respond(w, r, err.Error(), http.StatusInternalServerError)
@@ -340,7 +357,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 		log.Debug("new manifest", "ruid", r.ruid, "key", addr)
 	}
 
-	newAddr, err := s.api.UpdateManifest(r.Context(), addr, func(mw *api.ManifestWriter) error {
+	newAddr, err := s.api.UpdateManifest(ctx, addr, func(mw *api.ManifestWriter) error {
 		switch contentType {
 
 		case "application/x-tar":
@@ -509,6 +526,14 @@ func resourcePostMode(path string) (isRaw bool, frequency uint64, err error) {
 // and name "foo.eth" will be created
 func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 	log.Debug("handle.post.resource", "ruid", r.ruid)
+
+	var sp opentracing.Span
+	ctx := r.Context()
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"http.post.resource")
+	defer sp.Finish()
+
 	var err error
 	var addr storage.Address
 	var name string
@@ -525,7 +550,7 @@ func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 		name = r.uri.Addr
 
 		// the key is the content addressed root chunk holding mutable resource metadata information
-		addr, err = s.api.ResourceCreate(r.Context(), name, frequency)
+		addr, err = s.api.ResourceCreate(ctx, name, frequency)
 		if err != nil {
 			code, err2 := s.translateResourceError(w, r, "resource creation fail", err)
 
@@ -536,7 +561,7 @@ func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 		// we create a manifest so we can retrieve the resource with bzz:// later
 		// this manifest has a special "resource type" manifest, and its hash is the key of the mutable resource
 		// root chunk
-		m, err := s.api.NewResourceManifest(r.Context(), addr.Hex())
+		m, err := s.api.NewResourceManifest(ctx, addr.Hex())
 		if err != nil {
 			Respond(w, r, fmt.Sprintf("failed to create resource manifest: %v", err), http.StatusInternalServerError)
 			return
@@ -556,7 +581,7 @@ func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 		// that means that we retrieve the manifest and inspect its Hash member.
 		manifestAddr := r.uri.Address()
 		if manifestAddr == nil {
-			manifestAddr, err = s.api.Resolve(r.Context(), r.uri)
+			manifestAddr, err = s.api.Resolve(ctx, r.uri)
 			if err != nil {
 				getFail.Inc(1)
 				Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
@@ -567,7 +592,7 @@ func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 		}
 
 		// get the root chunk key from the manifest
-		addr, err = s.api.ResolveResourceManifest(r.Context(), manifestAddr)
+		addr, err = s.api.ResolveResourceManifest(ctx, manifestAddr)
 		if err != nil {
 			getFail.Inc(1)
 			Respond(w, r, fmt.Sprintf("error resolving resource root chunk for %s: %s", r.uri.Addr, err), http.StatusNotFound)
@@ -576,7 +601,7 @@ func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 
 		log.Debug("handle.post.resource: resolved", "ruid", r.ruid, "manifestkey", manifestAddr, "rootchunkkey", addr)
 
-		name, _, err = s.api.ResourceLookup(r.Context(), addr, 0, 0, &mru.LookupParams{})
+		name, _, err = s.api.ResourceLookup(ctx, addr, 0, 0, &mru.LookupParams{})
 		if err != nil {
 			Respond(w, r, err.Error(), http.StatusNotFound)
 			return
@@ -592,7 +617,7 @@ func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 
 	// Multihash will be passed as hex-encoded data, so we need to parse this to bytes
 	if isRaw {
-		_, _, _, err = s.api.ResourceUpdate(r.Context(), name, data)
+		_, _, _, err = s.api.ResourceUpdate(ctx, name, data)
 		if err != nil {
 			Respond(w, r, err.Error(), http.StatusBadRequest)
 			return
@@ -603,7 +628,7 @@ func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
 			Respond(w, r, err.Error(), http.StatusBadRequest)
 			return
 		}
-		_, _, _, err = s.api.ResourceUpdateMultihash(r.Context(), name, bytesdata)
+		_, _, _, err = s.api.ResourceUpdateMultihash(ctx, name, bytesdata)
 		if err != nil {
 			Respond(w, r, err.Error(), http.StatusBadRequest)
 			return
@@ -730,10 +755,18 @@ func (s *Server) translateResourceError(w http.ResponseWriter, r *Request, supEr
 func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
 	log.Debug("handle.get", "ruid", r.ruid, "uri", r.uri)
 	getCount.Inc(1)
+
+	var sp opentracing.Span
+	ctx := r.Context()
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"http.get")
+	defer sp.Finish()
+
 	var err error
 	addr := r.uri.Address()
 	if addr == nil {
-		addr, err = s.api.Resolve(r.Context(), r.uri)
+		addr, err = s.api.Resolve(ctx, r.uri)
 		if err != nil {
 			getFail.Inc(1)
 			Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
@@ -748,7 +781,7 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
 	// if path is set, interpret <key> as a manifest and return the
 	// raw entry at the given path
 	if r.uri.Path != "" {
-		walker, err := s.api.NewManifestWalker(r.Context(), addr, nil)
+		walker, err := s.api.NewManifestWalker(ctx, addr, nil)
 		if err != nil {
 			getFail.Inc(1)
 			Respond(w, r, fmt.Sprintf("%s is not a manifest", addr), http.StatusBadRequest)
@@ -796,8 +829,8 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
 	}
 
 	// check the root chunk exists by retrieving the file's size
-	reader, isEncrypted := s.api.Retrieve(r.Context(), addr)
-	if _, err := reader.Size(nil); err != nil {
+	reader, isEncrypted := s.api.Retrieve(ctx, addr)
+	if _, err := reader.Size(ctx, nil); err != nil {
 		getFail.Inc(1)
 		Respond(w, r, fmt.Sprintf("root chunk not found %s: %s", addr, err), http.StatusNotFound)
 		return
@@ -828,13 +861,21 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
 func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
 	log.Debug("handle.get.list", "ruid", r.ruid, "uri", r.uri)
 	getListCount.Inc(1)
+
+	var sp opentracing.Span
+	ctx := r.Context()
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"http.get.list")
+	defer sp.Finish()
+
 	// ensure the root path has a trailing slash so that relative URLs work
 	if r.uri.Path == "" && !strings.HasSuffix(r.URL.Path, "/") {
 		http.Redirect(w, &r.Request, r.URL.Path+"/", http.StatusMovedPermanently)
 		return
 	}
 
-	addr, err := s.api.Resolve(r.Context(), r.uri)
+	addr, err := s.api.Resolve(ctx, r.uri)
 	if err != nil {
 		getListFail.Inc(1)
 		Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
@@ -842,7 +883,7 @@ func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
 	}
 	log.Debug("handle.get.list: resolved", "ruid", r.ruid, "key", addr)
 
-	list, err := s.api.GetManifestList(r.Context(), addr, r.uri.Path)
+	list, err := s.api.GetManifestList(ctx, addr, r.uri.Path)
 	if err != nil {
 		getListFail.Inc(1)
 		Respond(w, r, err.Error(), http.StatusInternalServerError)
@@ -877,19 +918,28 @@ func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
 func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
 	log.Debug("handle.get.file", "ruid", r.ruid)
 	getFileCount.Inc(1)
+
+	var sp opentracing.Span
+	ctx := r.Context()
+	ctx, sp = spancontext.StartSpan(
+		ctx,
+		"http.get.file")
+
 	// ensure the root path has a trailing slash so that relative URLs work
 	if r.uri.Path == "" && !strings.HasSuffix(r.URL.Path, "/") {
 		http.Redirect(w, &r.Request, r.URL.Path+"/", http.StatusMovedPermanently)
+		sp.Finish()
 		return
 	}
 	var err error
 	manifestAddr := r.uri.Address()
 
 	if manifestAddr == nil {
-		manifestAddr, err = s.api.Resolve(r.Context(), r.uri)
+		manifestAddr, err = s.api.Resolve(ctx, r.uri)
 		if err != nil {
 			getFileFail.Inc(1)
 			Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
+			sp.Finish()
 			return
 		}
 	} else {
@@ -897,7 +947,8 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
 	}
 
 	log.Debug("handle.get.file: resolved", "ruid", r.ruid, "key", manifestAddr)
-	reader, contentType, status, contentKey, err := s.api.Get(r.Context(), manifestAddr, r.uri.Path)
+
+	reader, contentType, status, contentKey, err := s.api.Get(ctx, manifestAddr, r.uri.Path)
 
 	etag := common.Bytes2Hex(contentKey)
 	noneMatchEtag := r.Header.Get("If-None-Match")
@@ -905,6 +956,7 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
 	if noneMatchEtag != "" {
 		if bytes.Equal(storage.Address(common.Hex2Bytes(noneMatchEtag)), contentKey) {
 			Respond(w, r, "Not Modified", http.StatusNotModified)
+			sp.Finish()
 			return
 		}
 	}
@@ -918,34 +970,49 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
 			getFileFail.Inc(1)
 			Respond(w, r, err.Error(), http.StatusInternalServerError)
 		}
+		sp.Finish()
 		return
 	}
 
 	//the request results in ambiguous files
 	//e.g. /read with readme.md and readinglist.txt available in manifest
 	if status == http.StatusMultipleChoices {
-		list, err := s.api.GetManifestList(r.Context(), manifestAddr, r.uri.Path)
+		list, err := s.api.GetManifestList(ctx, manifestAddr, r.uri.Path)
 		if err != nil {
 			getFileFail.Inc(1)
 			Respond(w, r, err.Error(), http.StatusInternalServerError)
+			sp.Finish()
 			return
 		}
 
 		log.Debug(fmt.Sprintf("Multiple choices! --> %v", list), "ruid", r.ruid)
 		//show a nice page links to available entries
 		ShowMultipleChoices(w, r, list)
+		sp.Finish()
 		return
 	}
 
 	// check the root chunk exists by retrieving the file's size
-	if _, err := reader.Size(nil); err != nil {
+	if _, err := reader.Size(ctx, nil); err != nil {
 		getFileNotFound.Inc(1)
 		Respond(w, r, fmt.Sprintf("file not found %s: %s", r.uri, err), http.StatusNotFound)
+		sp.Finish()
 		return
 	}
 
+	buf, err := ioutil.ReadAll(newBufferedReadSeeker(reader, getFileBufferSize))
+	if err != nil {
+		getFileNotFound.Inc(1)
+		Respond(w, r, fmt.Sprintf("file not found %s: %s", r.uri, err), http.StatusNotFound)
+		sp.Finish()
+		return
+	}
+
+	log.Debug("got response in buffer", "len", len(buf), "ruid", r.ruid)
+	sp.Finish()
+
 	w.Header().Set("Content-Type", contentType)
-	http.ServeContent(w, &r.Request, "", time.Now(), newBufferedReadSeeker(reader, getFileBufferSize))
+	http.ServeContent(w, &r.Request, "", time.Now(), bytes.NewReader(buf))
 }
 
 // The size of buffer used for bufio.Reader on LazyChunkReader passed to
diff --git a/swarm/api/manifest.go b/swarm/api/manifest.go
index 78d1418bc..198ca22ce 100644
--- a/swarm/api/manifest.go
+++ b/swarm/api/manifest.go
@@ -212,10 +212,10 @@ func loadManifest(ctx context.Context, fileStore *storage.FileStore, hash storag
 	return readManifest(manifestReader, hash, fileStore, isEncrypted, quitC)
 }
 
-func readManifest(manifestReader storage.LazySectionReader, hash storage.Address, fileStore *storage.FileStore, isEncrypted bool, quitC chan bool) (trie *manifestTrie, err error) { // non-recursive, subtrees are downloaded on-demand
+func readManifest(mr storage.LazySectionReader, hash storage.Address, fileStore *storage.FileStore, isEncrypted bool, quitC chan bool) (trie *manifestTrie, err error) { // non-recursive, subtrees are downloaded on-demand
 
 	// TODO check size for oversized manifests
-	size, err := manifestReader.Size(quitC)
+	size, err := mr.Size(mr.Context(), quitC)
 	if err != nil { // size == 0
 		// can't determine size means we don't have the root chunk
 		log.Trace("manifest not found", "key", hash)
@@ -228,7 +228,7 @@ func readManifest(manifestReader storage.LazySectionReader, hash storage.Address
 		return
 	}
 	manifestData := make([]byte, size)
-	read, err := manifestReader.Read(manifestData)
+	read, err := mr.Read(manifestData)
 	if int64(read) < size {
 		log.Trace("manifest not found", "key", hash)
 		if err == nil {
diff --git a/swarm/api/storage.go b/swarm/api/storage.go
index 8646dc41f..3b52301a0 100644
--- a/swarm/api/storage.go
+++ b/swarm/api/storage.go
@@ -72,7 +72,7 @@ func (s *Storage) Get(ctx context.Context, bzzpath string) (*Response, error) {
 		return nil, err
 	}
 	quitC := make(chan bool)
-	expsize, err := reader.Size(quitC)
+	expsize, err := reader.Size(ctx, quitC)
 	if err != nil {
 		return nil, err
 	}