Merge pull request #334 from ethersphere/improved_tracing

improved tracing

p2p/discover/table.go

@@ -480,16 +480,16 @@ func (tab *Table) doRevalidate(done chan<- struct{}) {
 	b := tab.buckets[bi]
 	if err == nil {
 		// The node responded, move it to the front.
-		log.Debug("Revalidated node", "b", bi, "id", last.ID)
+		log.Trace("Revalidated node", "b", bi, "id", last.ID)
 		b.bump(last)
 		return
 	}
 	// No reply received, pick a replacement or delete the node if there aren't
 	// any replacements.
 	if r := tab.replace(b, last); r != nil {
-		log.Debug("Replaced dead node", "b", bi, "id", last.ID, "ip", last.IP, "r", r.ID, "rip", r.IP)
+		log.Trace("Replaced dead node", "b", bi, "id", last.ID, "ip", last.IP, "r", r.ID, "rip", r.IP)
 	} else {
-		log.Debug("Removed dead node", "b", bi, "id", last.ID, "ip", last.IP)
+		log.Trace("Removed dead node", "b", bi, "id", last.ID, "ip", last.IP)
 	}
 }
 

p2p/server.go

@@ -594,13 +594,13 @@ running:
 			// This channel is used by AddPeer to add to the
 			// ephemeral static peer list. Add it to the dialer,
 			// it will keep the node connected.
-			srv.log.Debug("Adding static node", "node", n)
+			srv.log.Trace("Adding static node", "node", n)
 			dialstate.addStatic(n)
 		case n := <-srv.removestatic:
 			// This channel is used by RemovePeer to send a
 			// disconnect request to a peer and begin the
 			// stop keeping the node connected
-			srv.log.Debug("Removing static node", "node", n)
+			srv.log.Trace("Removing static node", "node", n)
 			dialstate.removeStatic(n)
 			if p, ok := peers[n.ID]; ok {
 				p.Disconnect(DiscRequested)

swarm/api/api.go

@@ -242,6 +242,7 @@ func (self *Api) Retrieve(key storage.Key) storage.LazySectionReader {
 }
 
 func (self *Api) Store(data io.Reader, size int64) (key storage.Key, wait func(), err error) {
+	log.Debug("api.store", "size", size)
 	return self.dpa.Store(data, size)
 }
 
@@ -250,7 +251,7 @@ type ErrResolve error
 // DNS Resolver
 func (self *Api) Resolve(uri *URI) (storage.Key, error) {
 	apiResolveCount.Inc(1)
-	log.Trace(fmt.Sprintf("Resolving : %v", uri.Addr))
+	log.Trace("resolving", "uri", uri.Addr)
 
 	// if the URI is immutable, check if the address is a hash
 	isHash := hashMatcher.MatchString(uri.Addr)
@@ -307,6 +308,7 @@ func (self *Api) Put(content, contentType string) (k storage.Key, wait func(), e
 // to resolve basePath to content using dpa retrieve
 // it returns a section reader, mimeType, status and an error
 func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionReader, mimeType string, status int, err error) {
+	log.Debug("api.get", "key", key, "path", path)
 	apiGetCount.Inc(1)
 	trie, err := loadManifest(self.dpa, key, nil)
 	if err != nil {
@@ -316,9 +318,9 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
 		return
 	}
 
-	log.Trace(fmt.Sprintf("getEntry(%s)", path))
+	log.Trace("trie getting entry", "key", key, "path", path)
-
 	entry, _ := trie.getEntry(path)
+	log.Trace("trie got entry", "key", key, "path", path)
 
 	if entry != nil {
 		// we want to be able to serve Mutable Resource Updates transparently using the bzz:// scheme
@@ -330,7 +332,7 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
 		// we return a typed error instead. Since for all other purposes this is an invalid manifest,
 		// any normal interfacing code will just see an error fail accordingly.
 		if entry.ContentType == ResourceContentType {
-			log.Warn("resource type", "hash", entry.Hash)
+			log.Warn("resource type", "key", key, "hash", entry.Hash)
 			return nil, entry.ContentType, http.StatusOK, &ErrResourceReturn{entry.Hash}
 		}
 		key = common.Hex2Bytes(entry.Hash)
@@ -340,14 +342,14 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
 			return
 		} else {
 			mimeType = entry.ContentType
-			log.Trace(fmt.Sprintf("content lookup key: '%v' (%v)", key, mimeType))
+			log.Trace("content lookup key", "key", key, "mimetype", mimeType)
 			reader = self.dpa.Retrieve(key)
 		}
 	} else {
 		status = http.StatusNotFound
 		apiGetNotFound.Inc(1)
 		err = fmt.Errorf("manifest entry for '%s' not found", path)
-		log.Warn(fmt.Sprintf("%v", err))
+		log.Trace("manifest entry not found", "key", key, "path", path)
 	}
 	return
 }

swarm/api/http/server.go

@@ -121,6 +121,8 @@ type Request struct {
 // HandlePostRaw handles a POST request to a raw bzz-raw:/ URI, stores the request
 // body in swarm and returns the resulting storage key as a text/plain response
 func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
+	log.Debug("handle.post.raw", "ruid", r.ruid)
+
 	postRawCount.Inc(1)
 	if r.uri.Path != "" {
 		postRawFail.Inc(1)
@@ -140,7 +142,8 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
 		Respond(w, r, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	log.Debug(fmt.Sprintf("content for %s stored", key.Log()), "ruid", r.ruid)
+
+	log.Debug("stored content", "ruid", r.ruid, "key", key)
 
 	w.Header().Set("Content-Type", "text/plain")
 	w.WriteHeader(http.StatusOK)
@@ -153,6 +156,8 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
 // existing manifest or to a new manifest under <path> and returns the
 // 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)
 	contentType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
 	if err != nil {
@@ -169,6 +174,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 			Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusInternalServerError)
 			return
 		}
+		log.Debug("resolved key", "ruid", r.ruid, "key", key)
 	} else {
 		key, err = s.api.NewManifest()
 		if err != nil {
@@ -176,6 +182,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 			Respond(w, r, err.Error(), http.StatusInternalServerError)
 			return
 		}
+		log.Debug("new manifest", "ruid", r.ruid, "key", key)
 	}
 
 	newKey, err := s.updateManifest(key, func(mw *api.ManifestWriter) error {
@@ -197,12 +204,15 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
 		return
 	}
 
+	log.Debug("stored content", "ruid", r.ruid, "key", newKey)
+
 	w.Header().Set("Content-Type", "text/plain")
 	w.WriteHeader(http.StatusOK)
 	fmt.Fprint(w, newKey)
 }
 
 func (s *Server) handleTarUpload(req *Request, mw *api.ManifestWriter) error {
+	log.Debug("handle.tar.upload", "ruid", req.ruid)
 	tr := tar.NewReader(req.Body)
 	for {
 		hdr, err := tr.Next()
@@ -226,16 +236,17 @@ func (s *Server) handleTarUpload(req *Request, mw *api.ManifestWriter) error {
 			Size:        hdr.Size,
 			ModTime:     hdr.ModTime,
 		}
-		log.Debug(fmt.Sprintf("adding %s (%d bytes) to new manifest", entry.Path, entry.Size))
+		log.Debug("adding path to new manifest", "ruid", req.ruid, "bytes", entry.Size, "path", entry.Path)
 		contentKey, err := mw.AddEntry(tr, entry)
 		if err != nil {
 			return fmt.Errorf("error adding manifest entry from tar stream: %s", err)
 		}
-		log.Debug(fmt.Sprintf("content for %s stored", contentKey.Log()))
+		log.Debug("stored content", "ruid", req.ruid, "key", contentKey)
 	}
 }
 
 func (s *Server) handleMultipartUpload(req *Request, boundary string, mw *api.ManifestWriter) error {
+	log.Debug("handle.multipart.upload", "ruid", req.ruid)
 	mr := multipart.NewReader(req.Body, boundary)
 	for {
 		part, err := mr.NextPart()
@@ -283,16 +294,17 @@ func (s *Server) handleMultipartUpload(req *Request, boundary string, mw *api.Ma
 			Size:        size,
 			ModTime:     time.Now(),
 		}
-		log.Debug(fmt.Sprintf("adding %s (%d bytes) to new manifest", entry.Path, entry.Size))
+		log.Debug("adding path to new manifest", "ruid", req.ruid, "bytes", entry.Size, "path", entry.Path)
 		contentKey, err := mw.AddEntry(reader, entry)
 		if err != nil {
 			return fmt.Errorf("error adding manifest entry from multipart form: %s", err)
 		}
-		log.Debug(fmt.Sprintf("content for %s stored", contentKey.Log()))
+		log.Debug("stored content", "ruid", req.ruid, "key", contentKey)
 	}
 }
 
 func (s *Server) handleDirectUpload(req *Request, mw *api.ManifestWriter) error {
+	log.Debug("handle.direct.upload", "ruid", req.ruid)
 	key, err := mw.AddEntry(req.Body, &api.ManifestEntry{
 		Path:        req.uri.Path,
 		ContentType: req.Header.Get("Content-Type"),
@@ -303,7 +315,7 @@ func (s *Server) handleDirectUpload(req *Request, mw *api.ManifestWriter) error
 	if err != nil {
 		return err
 	}
-	log.Debug(fmt.Sprintf("content for %s stored", key.Log()))
+	log.Debug("stored content", "ruid", req.ruid, "key", key)
 	return nil
 }
 
@@ -311,6 +323,8 @@ func (s *Server) handleDirectUpload(req *Request, mw *api.ManifestWriter) error
 // <path> from <manifest> and returns the resulting manifest hash as a
 // text/plain response
 func (s *Server) HandleDelete(w http.ResponseWriter, r *Request) {
+	log.Debug("handle.delete", "ruid", r.ruid)
+
 	deleteCount.Inc(1)
 	key, err := s.api.Resolve(r.uri)
 	if err != nil {
@@ -335,6 +349,8 @@ func (s *Server) HandleDelete(w http.ResponseWriter, r *Request) {
 }
 
 func (s *Server) HandlePostResource(w http.ResponseWriter, r *Request) {
+	log.Debug("handle.post.resource", "ruid", r.ruid)
+
 	var outdata []byte
 	if r.uri.Path != "" {
 		frequency, err := strconv.ParseUint(r.uri.Path, 10, 64)
@@ -398,6 +414,7 @@ func (s *Server) HandleGetResource(w http.ResponseWriter, r *Request) {
 }
 
 func (s *Server) handleGetResource(w http.ResponseWriter, r *Request, name string) {
+	log.Debug("handle.get.resource", "ruid", r.ruid)
 	var params []string
 	if len(r.uri.Path) > 0 {
 		params = strings.Split(r.uri.Path, "/")
@@ -470,6 +487,7 @@ func (s *Server) translateResourceError(w http.ResponseWriter, r *Request, supEr
 // - bzz-hash://<key> and responds with the hash of the content stored
 //   at the given storage key as a text/plain response
 func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
+	log.Debug("handle.get", "ruid", r.ruid, "uri", r.uri)
 	getCount.Inc(1)
 	key, err := s.api.Resolve(r.uri)
 	if err != nil {
@@ -477,6 +495,7 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
 		Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
 		return
 	}
+	log.Debug("handle.get: resolved", "ruid", r.ruid, "key", key)
 
 	// if path is set, interpret <key> as a manifest and return the
 	// raw entry at the given path
@@ -548,6 +567,7 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
 // header of "application/x-tar" and returns a tar stream of all files
 // contained in the manifest
 func (s *Server) HandleGetFiles(w http.ResponseWriter, r *Request) {
+	log.Debug("handle.get.files", "ruid", r.ruid, "uri", r.uri)
 	getFilesCount.Inc(1)
 	if r.uri.Path != "" {
 		getFilesFail.Inc(1)
@@ -561,6 +581,7 @@ func (s *Server) HandleGetFiles(w http.ResponseWriter, r *Request) {
 		Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
 		return
 	}
+	log.Debug("handle.get.files: resolved", "ruid", r.ruid, "key", key)
 
 	walker, err := s.api.NewManifestWalker(key, nil)
 	if err != nil {
@@ -621,6 +642,7 @@ func (s *Server) HandleGetFiles(w http.ResponseWriter, r *Request) {
 // a list of all files contained in <manifest> under <path> grouped into
 // common prefixes using "/" as a delimiter
 func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
+	log.Debug("handle.get.list", "ruid", r.ruid, "uri", r.uri)
 	getListCount.Inc(1)
 	// ensure the root path has a trailing slash so that relative URLs work
 	if r.uri.Path == "" && !strings.HasSuffix(r.URL.Path, "/") {
@@ -634,6 +656,7 @@ func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
 		Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
 		return
 	}
+	log.Debug("handle.get.list: resolved", "ruid", r.ruid, "key", key)
 
 	list, err := s.getManifestList(key, r.uri.Path)
 
@@ -725,6 +748,7 @@ func (s *Server) getManifestList(key storage.Key, prefix string) (list api.Manif
 // HandleGetFile handles a GET request to bzz://<manifest>/<path> and responds
 // with the content of the file at <path> from the given <manifest>
 func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
+	log.Debug("handle.get.file", "ruid", r.ruid)
 	getFileCount.Inc(1)
 	// ensure the root path has a trailing slash so that relative URLs work
 	if r.uri.Path == "" && !strings.HasSuffix(r.URL.Path, "/") {
@@ -738,6 +762,7 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
 		Respond(w, r, fmt.Sprintf("cannot resolve %s: %s", r.uri.Addr, err), http.StatusNotFound)
 		return
 	}
+	log.Debug("handle.get.file: resolved", "ruid", r.ruid, "key", key)
 
 	reader, contentType, status, err := s.api.Get(key, r.uri.Path)
 
@@ -788,10 +813,13 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
 	http.ServeContent(w, &r.Request, "", time.Now(), reader)
 }
 
-func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+func (s *Server) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
 	req := &Request{Request: *r, ruid: uuid.New()[:8]}
 	requestCount.Inc(1)
-	log.Info("serve request", "ruid", req.ruid, "method", r.Method, "url", r.RequestURI)
+	log.Info("serving request", "ruid", req.ruid, "method", r.Method, "url", r.RequestURI)
+
+	// wrapping the ResponseWriter, so that we get the response code set by http.ServeContent
+	w := newLoggingResponseWriter(rw)
 
 	if r.RequestURI == "/" && strings.Contains(r.Header.Get("Accept"), "text/html") {
 
@@ -869,6 +897,8 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	default:
 		Respond(w, req, fmt.Sprintf("%s method is not supported", r.Method), http.StatusMethodNotAllowed)
 	}
+
+	log.Info("served response", "ruid", req.ruid, "code", w.statusCode)
 }
 
 func (s *Server) updateManifest(key storage.Key, update func(mw *api.ManifestWriter) error) (storage.Key, error) {
@@ -888,3 +918,17 @@ func (s *Server) updateManifest(key storage.Key, update func(mw *api.ManifestWri
 	log.Debug(fmt.Sprintf("generated manifest %s", key))
 	return key, nil
 }
+
+type loggingResponseWriter struct {
+	http.ResponseWriter
+	statusCode int
+}
+
+func newLoggingResponseWriter(w http.ResponseWriter) *loggingResponseWriter {
+	return &loggingResponseWriter{w, http.StatusOK}
+}
+
+func (lrw *loggingResponseWriter) WriteHeader(code int) {
+	lrw.statusCode = code
+	lrw.ResponseWriter.WriteHeader(code)
+}

swarm/api/manifest.go

@@ -201,10 +201,10 @@ type manifestTrieEntry struct {
 }
 
 func loadManifest(dpa *storage.DPA, hash storage.Key, quitC chan bool) (trie *manifestTrie, err error) { // non-recursive, subtrees are downloaded on-demand
-
+	log.Trace("manifest lookup", "key", hash)
-	log.Trace(fmt.Sprintf("manifest lookup key: '%v'.", hash.Log()))
 	// retrieve manifest via DPA
 	manifestReader := dpa.Retrieve(hash)
+	log.Trace("reader retrieved", "key", hash)
 	return readManifest(manifestReader, hash, dpa, quitC)
 }
 
@@ -214,31 +214,32 @@ func readManifest(manifestReader storage.LazySectionReader, hash storage.Key, dp
 	size, err := manifestReader.Size(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)
 		err = fmt.Errorf("Manifest not Found")
 		return
 	}
 	manifestData := make([]byte, size)
 	read, err := manifestReader.Read(manifestData)
 	if int64(read) < size {
-		log.Trace(fmt.Sprintf("Manifest %v not found.", hash.Log()))
+		log.Trace("manifest not found", "key", hash)
 		if err == nil {
 			err = fmt.Errorf("Manifest retrieval cut short: read %v, expect %v", read, size)
 		}
 		return
 	}
 
-	log.Trace(fmt.Sprintf("Manifest %v retrieved", hash.Log()))
+	log.Trace("manifest retrieved", "key", hash)
 	var man struct {
 		Entries []*manifestTrieEntry `json:"entries"`
 	}
 	err = json.Unmarshal(manifestData, &man)
 	if err != nil {
 		err = fmt.Errorf("Manifest %v is malformed: %v", hash.Log(), err)
-		log.Trace(fmt.Sprintf("%v", err))
+		log.Trace("malformed manifest", "key", hash)
 		return
 	}
 
-	log.Trace(fmt.Sprintf("Manifest %v has %d entries.", hash.Log(), len(man.Entries)))
+	log.Trace("manifest entries", "key", hash, "len", len(man.Entries))
 
 	trie = &manifestTrie{
 		dpa: dpa,
@@ -435,7 +436,6 @@ func (self *manifestTrie) listWithPrefix(prefix string, quitC chan bool, cb func
 }
 
 func (self *manifestTrie) findPrefixOf(path string, quitC chan bool) (entry *manifestTrieEntry, pos int) {
-
 	log.Trace(fmt.Sprintf("findPrefixOf(%s)", path))
 
 	if len(path) == 0 {

swarm/storage/memstore.go

@@ -21,6 +21,7 @@ package storage
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/metrics"
@@ -252,6 +253,8 @@ func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) {
 }
 
 func (s *MemStore) removeOldest() {
+	defer metrics.GetOrRegisterResettingTimer("memstore.purge", metrics.DefaultRegistry).UpdateSince(time.Now())
+
 	node := s.memtree
 	log.Warn("purge memstore")
 	for node.entry == nil {

swarm/storage/pyramid.go

@@ -169,7 +169,7 @@ func (self *PyramidChunker) decrementWorkerCount() {
 }
 
 func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk) (k Key, wait func(), err error) {
-	log.Trace("pyramid.chunker: Split()")
+	log.Debug("pyramid.chunker: Split()", "size", size)
 	jobC := make(chan *chunkJob, 2*ChunkProcessors)
 	wg := &sync.WaitGroup{}
 	storageWG := &sync.WaitGroup{}
@@ -204,11 +204,10 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk
 	case <-time.NewTimer(splitTimeout).C:
 	}
 	return rootKey, storageWG.Wait, nil
-
 }
 
 func (self *PyramidChunker) Append(key Key, data io.Reader, chunkC chan *Chunk) (k Key, wait func(), err error) {
-	log.Trace("pyramid.chunker: Append()")
+	log.Debug("pyramid.chunker: Append()")
 	quitC := make(chan bool)
 	rootKey := make([]byte, self.hashSize)
 	chunkLevel := make([][]*TreeEntry, self.branches)
@@ -267,7 +266,7 @@ func (self *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan
 }
 
 func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJob, chunkC chan *Chunk, storageWG *sync.WaitGroup) {
-	log.Trace("pyramid.chunker: processChunk()", "id", id)
+	log.Debug("pyramid.chunker: processChunk()", "id", id)
 
 	hasher.ResetWithLength(job.chunk[:8]) // 8 bytes of length
 	hasher.Write(job.chunk[8:])           // minus 8 []byte length
@@ -294,7 +293,7 @@ func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJ
 }
 
 func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC chan *Chunk, quitC chan bool) error {
-	log.Trace("pyramid.chunker: loadTree()")
+	log.Debug("pyramid.chunker: loadTree()")
 	// Get the root chunk to get the total size
 	chunk := retrieve(key, chunkC, quitC)
 	if chunk == nil {
@@ -377,7 +376,7 @@ func (self *PyramidChunker) loadTree(chunkLevel [][]*TreeEntry, key Key, chunkC
 }
 
 func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEntry, data io.Reader, rootKey []byte, quitC chan bool, wg *sync.WaitGroup, jobC chan *chunkJob, chunkC chan *Chunk, errC chan error, storageWG *sync.WaitGroup) {
-	log.Trace("pyramid.chunker: prepareChunks", "isAppend", isAppend)
+	log.Debug("pyramid.chunker: prepareChunks", "isAppend", isAppend)
 	defer wg.Done()
 
 	chunkWG := &sync.WaitGroup{}