forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

swarm: initial instrumentation with go-metrics

Browse source
This commit is contained in:
Fabio Barone 2018-01-12 14:42:07 -05:00 committed by Anton Evangelatov
parent 44d40ffce1
commit 89dfc96ec3
14 changed files with 289 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/swarm/main.go
View file

@ -334,6 +334,7 @@ DEPRECATED: use 'swarm db clean'.
utils.IPCDisabledFlag,
utils.IPCPathFlag,
utils.PasswordFileFlag,
utils.MetricsEnabledFlag,
// bzzd-specific flags
CorsStringFlag,
EnsAPIFlag,

9
metrics/metrics.go
View file

@ -38,6 +38,15 @@ func init() {
//exp.Exp(DefaultRegistry)
}
// NewGauge creates a new metrics Gauge, either a real one of a NOP stub depending
// on the metrics flag.
func NewGauge(name string) metrics.Gauge {
if !Enabled {
return new(metrics.NilGauge)
}
return metrics.GetOrRegisterGauge(name, metrics.DefaultRegistry)
}
// CollectProcessMetrics periodically collects various metrics about the running
// process.
func CollectProcessMetrics(refresh time.Duration) {

65
swarm/api/api.go
View file

@ -32,11 +32,33 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/swarm/storage"
)
var hashMatcher = regexp.MustCompile("^[0-9A-Fa-f]{64}")
//setup metrics
var (
apiResolveCount = metrics.NewCounter("api.resolve.count")
apiResolveFail = metrics.NewCounter("api.resolve.fail")
apiPutCount = metrics.NewCounter("api.put.count")
apiPutFail = metrics.NewCounter("api.put.fail")
apiGetCount = metrics.NewCounter("api.get.count")
apiGetNotFound = metrics.NewCounter("api.get.notfound")
apiGetHttp300 = metrics.NewCounter("api.get.http300")
apiModifyCount = metrics.NewCounter("api.modify.count")
apiModifyFail = metrics.NewCounter("api.modify.fail")
apiAddFileCount = metrics.NewCounter("api.addfile.count")
apiAddFileFail = metrics.NewCounter("api.addfile.fail")
apiRmFileCount = metrics.NewCounter("api.removefile.count")
apiRmFileFail = metrics.NewCounter("api.removefile.fail")
apiAppendFileCount = metrics.NewCounter("api.appendfile.count")
apiAppendFileFail = metrics.NewCounter("api.appendfile.fail")
apiBuildDirTreeCount = metrics.NewCounter("api.builddirtree.fail")
apiBuildDirTreeFail = metrics.NewCounter("api.builddirtree.count")
)
type Resolver interface {
Resolve(string) (common.Hash, error)
}
@ -155,12 +177,14 @@ 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))
// if the URI is immutable, check if the address is a hash
isHash := hashMatcher.MatchString(uri.Addr)
if uri.Immutable() || uri.DeprecatedImmutable() {
if !isHash {
apiResolveFail.Inc(1)
return nil, fmt.Errorf("immutable address not a content hash: %q", uri.Addr)
}
return common.Hex2Bytes(uri.Addr), nil
@ -169,6 +193,7 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
// if DNS is not configured, check if the address is a hash
if self.dns == nil {
if !isHash {
apiResolveFail.Inc(1)
return nil, fmt.Errorf("no DNS to resolve name: %q", uri.Addr)
}
return common.Hex2Bytes(uri.Addr), nil
@ -179,6 +204,7 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
if err == nil {
return resolved[:], nil
} else if !isHash {
apiResolveFail.Inc(1)
return nil, err
}
return common.Hex2Bytes(uri.Addr), nil
@ -186,16 +212,19 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
// Put provides singleton manifest creation on top of dpa store
func (self *Api) Put(content, contentType string) (storage.Key, error) {
apiPutCount.Inc(1)
r := strings.NewReader(content)
wg := &sync.WaitGroup{}
key, err := self.dpa.Store(r, int64(len(content)), wg, nil)
if err != nil {
apiPutFail.Inc(1)
return nil, err
}
manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType)
r = strings.NewReader(manifest)
key, err = self.dpa.Store(r, int64(len(manifest)), wg, nil)
if err != nil {
apiPutFail.Inc(1)
return nil, err
}
wg.Wait()
@ -206,8 +235,10 @@ func (self *Api) Put(content, contentType string) (storage.Key, error) {
// 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) {
apiGetCount.Inc(1)
trie, err := loadManifest(self.dpa, key, nil)
if err != nil {
apiGetNotFound.Inc(1)
status = http.StatusNotFound
log.Warn(fmt.Sprintf("loadManifestTrie error: %v", err))
return
@ -221,6 +252,7 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
key = common.Hex2Bytes(entry.Hash)
status = entry.Status
if status == http.StatusMultipleChoices {
apiGetHttp300.Inc(1)
return
} else {
mimeType = entry.ContentType
@ -229,6 +261,7 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
}
} else {
status = http.StatusNotFound
apiGetNotFound.Inc(1)
err = fmt.Errorf("manifest entry for '%s' not found", path)
log.Warn(fmt.Sprintf("%v", err))
}
@ -236,9 +269,11 @@ func (self *Api) Get(key storage.Key, path string) (reader storage.LazySectionRe
}
func (self *Api) Modify(key storage.Key, path, contentHash, contentType string) (storage.Key, error) {
apiModifyCount.Inc(1)
quitC := make(chan bool)
trie, err := loadManifest(self.dpa, key, quitC)
if err != nil {
apiModifyFail.Inc(1)
return nil, err
}
if contentHash != "" {
@ -253,6 +288,7 @@ func (self *Api) Modify(key storage.Key, path, contentHash, contentType string)
}
if err := trie.recalcAndStore(); err != nil {
apiModifyFail.Inc(1)
return nil, err
}
return trie.hash, nil
@ -260,12 +296,16 @@ func (self *Api) Modify(key storage.Key, path, contentHash, contentType string)
func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver bool) (storage.Key, string, error) {
apiAddFileCount.Inc(1)
uri, err := Parse("bzz:/" + mhash)
if err != nil {
apiAddFileFail.Inc(1)
return nil, "", err
}
mkey, err := self.Resolve(uri)
if err != nil {
apiAddFileFail.Inc(1)
return nil, "", err
}
@ -284,16 +324,19 @@ func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver
mw, err := self.NewManifestWriter(mkey, nil)
if err != nil {
apiAddFileFail.Inc(1)
return nil, "", err
}
fkey, err := mw.AddEntry(bytes.NewReader(content), entry)
if err != nil {
apiAddFileFail.Inc(1)
return nil, "", err
}
newMkey, err := mw.Store()
if err != nil {
apiAddFileFail.Inc(1)
return nil, "", err
}
@ -304,12 +347,16 @@ func (self *Api) AddFile(mhash, path, fname string, content []byte, nameresolver
func (self *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (string, error) {
apiRmFileCount.Inc(1)
uri, err := Parse("bzz:/" + mhash)
if err != nil {
apiRmFileFail.Inc(1)
return "", err
}
mkey, err := self.Resolve(uri)
if err != nil {
apiRmFileFail.Inc(1)
return "", err
}
@ -320,16 +367,19 @@ func (self *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (strin
mw, err := self.NewManifestWriter(mkey, nil)
if err != nil {
apiRmFileFail.Inc(1)
return "", err
}
err = mw.RemoveEntry(filepath.Join(path, fname))
if err != nil {
apiRmFileFail.Inc(1)
return "", err
}
newMkey, err := mw.Store()
if err != nil {
apiRmFileFail.Inc(1)
return "", err
}
@ -339,6 +389,8 @@ func (self *Api) RemoveFile(mhash, path, fname string, nameresolver bool) (strin
func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, content []byte, oldKey storage.Key, offset int64, addSize int64, nameresolver bool) (storage.Key, string, error) {
apiAppendFileCount.Inc(1)
buffSize := offset + addSize
if buffSize < existingSize {
buffSize = existingSize
@ -366,10 +418,12 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
uri, err := Parse("bzz:/" + mhash)
if err != nil {
apiAppendFileFail.Inc(1)
return nil, "", err
}
mkey, err := self.Resolve(uri)
if err != nil {
apiAppendFileFail.Inc(1)
return nil, "", err
}
@ -380,11 +434,13 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
mw, err := self.NewManifestWriter(mkey, nil)
if err != nil {
apiAppendFileFail.Inc(1)
return nil, "", err
}
err = mw.RemoveEntry(filepath.Join(path, fname))
if err != nil {
apiAppendFileFail.Inc(1)
return nil, "", err
}
@ -398,11 +454,13 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
fkey, err := mw.AddEntry(io.Reader(combinedReader), entry)
if err != nil {
apiAppendFileFail.Inc(1)
return nil, "", err
}
newMkey, err := mw.Store()
if err != nil {
apiAppendFileFail.Inc(1)
return nil, "", err
}
@ -412,18 +470,24 @@ func (self *Api) AppendFile(mhash, path, fname string, existingSize int64, conte
}
func (self *Api) BuildDirectoryTree(mhash string, nameresolver bool) (key storage.Key, manifestEntryMap map[string]*manifestTrieEntry, err error) {
apiBuildDirTreeCount.Inc(1)
uri, err := Parse("bzz:/" + mhash)
if err != nil {
apiBuildDirTreeFail.Inc(1)
return nil, nil, err
}
key, err = self.Resolve(uri)
if err != nil {
apiBuildDirTreeFail.Inc(1)
return nil, nil, err
}
quitC := make(chan bool)
rootTrie, err := loadManifest(self.dpa, key, quitC)
if err != nil {
apiBuildDirTreeFail.Inc(1)
return nil, nil, fmt.Errorf("can't load manifest %v: %v", key.String(), err)
}
@ -433,6 +497,7 @@ func (self *Api) BuildDirectoryTree(mhash string, nameresolver bool) (key storag
})
if err != nil {
apiBuildDirTreeFail.Inc(1)
return nil, nil, fmt.Errorf("list with prefix failed %v: %v", key.String(), err)
}
return key, manifestEntryMap, nil

9
swarm/api/http/error.go
View file

@ -29,12 +29,19 @@ import (
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/swarm/api"
)
//templateMap holds a mapping of an HTTP error code to a template
var templateMap map[int]*template.Template
//metrics variables
var (
htmlCounter = metrics.NewCounter("api.http.errorpage.html.count")
jsonCounter = metrics.NewCounter("api.http.errorpage.json.count")
)
//parameters needed for formatting the correct HTML page
type ErrorParams struct {
Msg string
@ -132,6 +139,7 @@ func respond(w http.ResponseWriter, r *http.Request, params *ErrorParams) {
//return a HTML page
func respondHtml(w http.ResponseWriter, params *ErrorParams) {
htmlCounter.Inc(1)
err := params.template.Execute(w, params)
if err != nil {
log.Error(err.Error())
@ -140,6 +148,7 @@ func respondHtml(w http.ResponseWriter, params *ErrorParams) {
//return JSON
func respondJson(w http.ResponseWriter, params *ErrorParams) {
jsonCounter.Inc(1)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(params)
}

57
swarm/api/http/server.go
View file

@ -37,11 +37,33 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/swarm/api"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/rs/cors"
)
//setup metrics
var (
postRawCount = metrics.NewCounter("api.http.post.raw.count")
postRawFail = metrics.NewCounter("api.http.post.raw.fail")
postFilesCount = metrics.NewCounter("api.http.post.files.count")
postFilesFail = metrics.NewCounter("api.http.post.files.fail")
deleteCount = metrics.NewCounter("api.http.delete.count")
deleteFail = metrics.NewCounter("api.http.delete.fail")
getCount = metrics.NewCounter("api.http.get.count")
getFail = metrics.NewCounter("api.http.get.fail")
getFileCount = metrics.NewCounter("api.http.get.file.count")
getFileNotFound = metrics.NewCounter("api.http.get.file.notfound")
getFileFail = metrics.NewCounter("api.http.get.file.fail")
getFilesCount = metrics.NewCounter("api.http.get.files.count")
getFilesFail = metrics.NewCounter("api.http.get.files.fail")
getListCount = metrics.NewCounter("api.http.get.list.count")
getListFail = metrics.NewCounter("api.http.get.list.fail")
requestCount = metrics.NewCounter("http.request.count")
requestTimer = metrics.NewTimer("http.request.time")
)
// ServerConfig is the basic configuration needed for the HTTP server and also
// includes CORS settings.
type ServerConfig struct {
@ -89,18 +111,22 @@ 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) {
postRawCount.Inc(1)
if r.uri.Path != "" {
postRawFail.Inc(1)
s.BadRequest(w, r, "raw POST request cannot contain a path")
return
}
if r.Header.Get("Content-Length") == "" {
postRawFail.Inc(1)
s.BadRequest(w, r, "missing Content-Length header in request")
return
}
key, err := s.api.Store(r.Body, r.ContentLength, nil)
if err != nil {
postRawFail.Inc(1)
s.Error(w, r, err)
return
}
@ -117,8 +143,10 @@ 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) {
postFilesCount.Inc(1)
contentType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
if err != nil {
postFilesFail.Inc(1)
s.BadRequest(w, r, err.Error())
return
}
@ -127,12 +155,14 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
if r.uri.Addr != "" {
key, err = s.api.Resolve(r.uri)
if err != nil {
postFilesFail.Inc(1)
s.Error(w, r, fmt.Errorf("error resolving %s: %s", r.uri.Addr, err))
return
}
} else {
key, err = s.api.NewManifest()
if err != nil {
postFilesFail.Inc(1)
s.Error(w, r, err)
return
}
@ -152,6 +182,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *Request) {
}
})
if err != nil {
postFilesFail.Inc(1)
s.Error(w, r, fmt.Errorf("error creating manifest: %s", err))
return
}
@ -270,8 +301,10 @@ 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) {
deleteCount.Inc(1)
key, err := s.api.Resolve(r.uri)
if err != nil {
deleteFail.Inc(1)
s.Error(w, r, fmt.Errorf("error resolving %s: %s", r.uri.Addr, err))
return
}
@ -281,6 +314,7 @@ func (s *Server) HandleDelete(w http.ResponseWriter, r *Request) {
return mw.RemoveEntry(r.uri.Path)
})
if err != nil {
deleteFail.Inc(1)
s.Error(w, r, fmt.Errorf("error updating manifest: %s", err))
return
}
@ -296,8 +330,10 @@ func (s *Server) HandleDelete(w http.ResponseWriter, r *Request) {
// - 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) {
getCount.Inc(1)
key, err := s.api.Resolve(r.uri)
if err != nil {
getFail.Inc(1)
s.Error(w, r, fmt.Errorf("error resolving %s: %s", r.uri.Addr, err))
return
}
@ -307,6 +343,7 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
if r.uri.Path != "" {
walker, err := s.api.NewManifestWalker(key, nil)
if err != nil {
getFail.Inc(1)
s.BadRequest(w, r, fmt.Sprintf("%s is not a manifest", key))
return
}
@ -335,6 +372,7 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
return api.SkipManifest
})
if entry == nil {
getFail.Inc(1)
s.NotFound(w, r, fmt.Errorf("Manifest entry could not be loaded"))
return
}
@ -344,6 +382,7 @@ func (s *Server) HandleGet(w http.ResponseWriter, r *Request) {
// check the root chunk exists by retrieving the file's size
reader := s.api.Retrieve(key)
if _, err := reader.Size(nil); err != nil {
getFail.Inc(1)
s.NotFound(w, r, fmt.Errorf("Root chunk not found %s: %s", key, err))
return
}
@ -370,19 +409,23 @@ 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) {
getFilesCount.Inc(1)
if r.uri.Path != "" {
getFilesFail.Inc(1)
s.BadRequest(w, r, "files request cannot contain a path")
return
}
key, err := s.api.Resolve(r.uri)
if err != nil {
getFilesFail.Inc(1)
s.Error(w, r, fmt.Errorf("error resolving %s: %s", r.uri.Addr, err))
return
}
walker, err := s.api.NewManifestWalker(key, nil)
if err != nil {
getFilesFail.Inc(1)
s.Error(w, r, err)
return
}
@ -430,6 +473,7 @@ func (s *Server) HandleGetFiles(w http.ResponseWriter, r *Request) {
return nil
})
if err != nil {
getFilesFail.Inc(1)
s.logError("error generating tar stream: %s", err)
}
}
@ -438,6 +482,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) {
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, "/") {
http.Redirect(w, &r.Request, r.URL.Path+"/", http.StatusMovedPermanently)
@ -446,6 +491,7 @@ func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
key, err := s.api.Resolve(r.uri)
if err != nil {
getListFail.Inc(1)
s.Error(w, r, fmt.Errorf("error resolving %s: %s", r.uri.Addr, err))
return
}
@ -453,6 +499,7 @@ func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
list, err := s.getManifestList(key, r.uri.Path)
if err != nil {
getListFail.Inc(1)
s.Error(w, r, err)
return
}
@ -470,6 +517,7 @@ func (s *Server) HandleGetList(w http.ResponseWriter, r *Request) {
List: &list,
})
if err != nil {
getListFail.Inc(1)
s.logError("error rendering list HTML: %s", err)
}
return
@ -538,6 +586,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) {
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, "/") {
http.Redirect(w, &r.Request, r.URL.Path+"/", http.StatusMovedPermanently)
@ -546,6 +595,7 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
key, err := s.api.Resolve(r.uri)
if err != nil {
getFileFail.Inc(1)
s.Error(w, r, fmt.Errorf("error resolving %s: %s", r.uri.Addr, err))
return
}
@ -554,8 +604,10 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
if err != nil {
switch status {
case http.StatusNotFound:
getFileNotFound.Inc(1)
s.NotFound(w, r, err)
default:
getFileFail.Inc(1)
s.Error(w, r, err)
}
return
@ -567,6 +619,7 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
list, err := s.getManifestList(key, r.uri.Path)
if err != nil {
getFileFail.Inc(1)
s.Error(w, r, err)
return
}
@ -579,6 +632,7 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
// check the root chunk exists by retrieving the file's size
if _, err := reader.Size(nil); err != nil {
getFileNotFound.Inc(1)
s.NotFound(w, r, fmt.Errorf("File not found %s: %s", r.uri, err))
return
}
@ -589,6 +643,9 @@ func (s *Server) HandleGetFile(w http.ResponseWriter, r *Request) {
}
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
requestCount.Inc(1)
startTime := time.Now()
defer requestTimer.UpdateSince(startTime)
s.logDebug("HTTP %s request URL: '%s', Host: '%s', Path: '%s', Referer: '%s', Accept: '%s'", r.Method, r.RequestURI, r.URL.Host, r.URL.Path, r.Referer(), r.Header.Get("Accept"))
uri, err := api.Parse(strings.TrimLeft(r.URL.Path, "/"))

15
swarm/network/depo.go
View file

@ -23,9 +23,19 @@ import (
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/swarm/storage"
)
//metrics variables
var (
syncReceiveCount = metrics.NewCounter("network.sync.recv.count")
syncReceiveIgnore = metrics.NewCounter("network.sync.recv.ignore")
syncSendCount = metrics.NewCounter("network.sync.send.count")
syncSendRefused = metrics.NewCounter("network.sync.send.refused")
syncSendNotFound = metrics.NewCounter("network.sync.send.notfound")
)
// Handler for storage/retrieval related protocol requests
// implements the StorageHandler interface used by the bzz protocol
type Depo struct {
@ -107,6 +117,7 @@ func (self *Depo) HandleStoreRequestMsg(req *storeRequestMsgData, p *peer) {
log.Trace(fmt.Sprintf("Depo.handleStoreRequest: %v not found locally. create new chunk/request", req.Key))
// not found in memory cache, ie., a genuine store request
// create chunk
syncReceiveCount.Inc(1)
chunk = storage.NewChunk(req.Key, nil)
case chunk.SData == nil:
@ -116,6 +127,7 @@ func (self *Depo) HandleStoreRequestMsg(req *storeRequestMsgData, p *peer) {
default:
// data is found, store request ignored
// this should update access count?
syncReceiveIgnore.Inc(1)
log.Trace(fmt.Sprintf("Depo.HandleStoreRequest: %v found locally. ignore.", req))
islocal = true
//return
@ -172,11 +184,14 @@ func (self *Depo) HandleRetrieveRequestMsg(req *retrieveRequestMsgData, p *peer)
SData: chunk.SData,
requestTimeout: req.timeout, //
}
syncSendCount.Inc(1)
p.syncer.addRequest(sreq, DeliverReq)
} else {
syncSendRefused.Inc(1)
log.Trace(fmt.Sprintf("Depo.HandleRetrieveRequest: %v - content found, not wanted", req.Key.Log()))
}
} else {
syncSendNotFound.Inc(1)
log.Trace(fmt.Sprintf("Depo.HandleRetrieveRequest: %v - content not found locally. asked swarm for help. will get back", req.Key.Log()))
}
}

10
swarm/network/hive.go
View file

@ -24,6 +24,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/netutil"
"github.com/ethereum/go-ethereum/swarm/network/kademlia"
@ -39,6 +40,12 @@ import (
// connections and disconnections are reported and relayed
// to keep the nodetable uptodate
var (
peersNumGauge = metrics.NewGauge("network.peers.num")
addPeerCounter = metrics.NewCounter("network.addpeer.count")
removePeerCounter = metrics.NewCounter("network.removepeer.count")
)
type Hive struct {
listenAddr func() string
callInterval uint64
@ -192,6 +199,7 @@ func (self *Hive) Start(id discover.NodeID, listenAddr func() string, connectPee
func (self *Hive) keepAlive() {
alarm := time.NewTicker(time.Duration(self.callInterval)).C
for {
peersNumGauge.Update(int64(self.kad.Count()))
select {
case <-alarm:
if self.kad.DBCount() > 0 {
@ -223,6 +231,7 @@ func (self *Hive) Stop() error {
// called at the end of a successful protocol handshake
func (self *Hive) addPeer(p *peer) error {
addPeerCounter.Inc(1)
defer func() {
select {
case self.more <- true:
@ -247,6 +256,7 @@ func (self *Hive) addPeer(p *peer) error {
// called after peer disconnected
func (self *Hive) removePeer(p *peer) {
removePeerCounter.Inc(1)
log.Debug(fmt.Sprintf("bee %v removed", p))
self.kad.Off(p, saveSync)
select {

9
swarm/network/kademlia/kademlia.go
View file

@ -24,6 +24,13 @@ import (
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
)
//metrics variables
var (
bucketOnIndexGauge = metrics.NewGauge("network.kademlia.bucket.onindex")
bucketOffIndexGauge = metrics.NewGauge("network.kademlia.bucket.offindex")
)
const (
@ -138,6 +145,7 @@ func (self *Kademlia) On(node Node, cb func(*NodeRecord, Node) error) (err error
// TODO: give priority to peers with active traffic
if len(bucket) < self.BucketSize { // >= allows us to add peers beyond the bucketsize limitation
self.buckets[index] = append(bucket, node)
bucketOnIndexGauge.Update(int64(index))
log.Debug(fmt.Sprintf("add node %v to table", node))
self.setProxLimit(index, true)
record.node = node
@ -178,6 +186,7 @@ func (self *Kademlia) Off(node Node, cb func(*NodeRecord, Node)) (err error) {
defer self.lock.Unlock()
index := self.proximityBin(node.Addr())
bucketOffIndexGauge.Update(int64(index))
bucket := self.buckets[index]
for i := 0; i < len(bucket); i++ {
if node.Addr() == bucket[i].Addr() {

24
swarm/network/protocol.go
View file

@ -39,12 +39,26 @@ import (
"github.com/ethereum/go-ethereum/contracts/chequebook"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/p2p"
bzzswap "github.com/ethereum/go-ethereum/swarm/services/swap"
"github.com/ethereum/go-ethereum/swarm/services/swap/swap"
"github.com/ethereum/go-ethereum/swarm/storage"
)
//metrics variables
var (
storeRequestMsgCounter = metrics.NewCounter("network.protocol.msg.storerequest.count")
retrieveRequestMsgCounter = metrics.NewCounter("network.protocol.msg.retrieverequest.count")
peersMsgCounter = metrics.NewCounter("network.protocol.msg.peers.count")
syncRequestMsgCounter = metrics.NewCounter("network.protocol.msg.syncrequest.count")
unsyncedKeysMsgCounter = metrics.NewCounter("network.protocol.msg.unsyncedkeys.count")
deliverRequestMsgCounter = metrics.NewCounter("network.protocol.msg.deliverrequest.count")
paymentMsgCounter = metrics.NewCounter("network.protocol.msg.payment.count")
invalidMsgCounter = metrics.NewCounter("network.protocol.msg.invalid.count")
handleStatusMsgCounter = metrics.NewCounter("network.protocol.msg.handlestatus.count")
)
const (
Version = 0
ProtocolLength = uint64(8)
@ -206,6 +220,7 @@ func (self *bzz) handle() error {
case storeRequestMsg:
// store requests are dispatched to netStore
storeRequestMsgCounter.Inc(1)
var req storeRequestMsgData
if err := msg.Decode(&req); err != nil {
return fmt.Errorf("<- %v: %v", msg, err)
@ -221,6 +236,7 @@ func (self *bzz) handle() error {
case retrieveRequestMsg:
// retrieve Requests are dispatched to netStore
retrieveRequestMsgCounter.Inc(1)
var req retrieveRequestMsgData
if err := msg.Decode(&req); err != nil {
return fmt.Errorf("<- %v: %v", msg, err)
@ -241,6 +257,7 @@ func (self *bzz) handle() error {
case peersMsg:
// response to lookups and immediate response to retrieve requests
// dispatches new peer data to the hive that adds them to KADDB
peersMsgCounter.Inc(1)
var req peersMsgData
if err := msg.Decode(&req); err != nil {
return fmt.Errorf("<- %v: %v", msg, err)
@ -250,6 +267,7 @@ func (self *bzz) handle() error {
self.hive.HandlePeersMsg(&req, &peer{bzz: self})
case syncRequestMsg:
syncRequestMsgCounter.Inc(1)
var req syncRequestMsgData
if err := msg.Decode(&req); err != nil {
return fmt.Errorf("<- %v: %v", msg, err)
@ -260,6 +278,7 @@ func (self *bzz) handle() error {
case unsyncedKeysMsg:
// coming from parent node offering
unsyncedKeysMsgCounter.Inc(1)
var req unsyncedKeysMsgData
if err := msg.Decode(&req); err != nil {
return fmt.Errorf("<- %v: %v", msg, err)
@ -274,6 +293,7 @@ func (self *bzz) handle() error {
case deliveryRequestMsg:
// response to syncKeysMsg hashes filtered not existing in db
// also relays the last synced state to the source
deliverRequestMsgCounter.Inc(1)
var req deliveryRequestMsgData
if err := msg.Decode(&req); err != nil {
return fmt.Errorf("<-msg %v: %v", msg, err)
@ -287,6 +307,7 @@ func (self *bzz) handle() error {
case paymentMsg:
// swap protocol message for payment, Units paid for, Cheque paid with
paymentMsgCounter.Inc(1)
if self.swapEnabled {
var req paymentMsgData
if err := msg.Decode(&req); err != nil {
@ -298,6 +319,7 @@ func (self *bzz) handle() error {
default:
// no other message is allowed
invalidMsgCounter.Inc(1)
return fmt.Errorf("invalid message code: %v", msg.Code)
}
return nil
@ -332,6 +354,8 @@ func (self *bzz) handleStatus() (err error) {
return fmt.Errorf("first msg has code %x (!= %x)", msg.Code, statusMsg)
}
handleStatusMsgCounter.Inc(1)
if msg.Size > ProtocolMaxMsgSize {
return fmt.Errorf("message too long: %v > %v", msg.Size, ProtocolMaxMsgSize)
}

14
swarm/storage/chunker.go
View file

@ -23,6 +23,8 @@ import (
"io"
"sync"
"time"
"github.com/ethereum/go-ethereum/metrics"
)
/*
@ -63,6 +65,11 @@ var (
errOperationTimedOut = errors.New("operation timed out")
)
//metrics variables
var (
newChunkCounter = metrics.NewCounter("storage.chunks.new")
)
type TreeChunker struct {
branches int64
hashFunc SwarmHasher
@ -298,6 +305,13 @@ func (self *TreeChunker) hashChunk(hasher SwarmHash, job *hashJob, chunkC chan *
job.parentWg.Done()
if chunkC != nil {
//NOTE: this increases the chunk count even if the local node already has this chunk;
//on file upload the node will increase this counter even if the same file has already been uploaded
//So it should be evaluated whether it is worth keeping this counter
//and/or actually better track when the chunk is Put to the local database
//(which may question the need for disambiguation when a completely new chunk has been created
//and/or a chunk is being put to the local DB; for chunk tracking it may be worth distinguishing
newChunkCounter.Inc(1)
chunkC <- newChunk
}
}

9
swarm/storage/dbstore.go
View file

@ -33,11 +33,18 @@ import (
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/rlp"
"github.com/syndtr/goleveldb/leveldb"
"github.com/syndtr/goleveldb/leveldb/iterator"
)
//metrics variables
var (
gcCounter = metrics.NewCounter("storage.db.dbstore.gc.count")
dbStoreDeleteCounter = metrics.NewCounter("storage.db.dbstore.rm.count")
)
const (
defaultDbCapacity = 5000000
defaultRadius = 0 // not yet used
@ -255,6 +262,7 @@ func (s *DbStore) collectGarbage(ratio float32) {
// actual gc
for i := 0; i < gcnt; i++ {
if s.gcArray[i].value <= cutval {
gcCounter.Inc(1)
s.delete(s.gcArray[i].idx, s.gcArray[i].idxKey)
}
}
@ -383,6 +391,7 @@ func (s *DbStore) delete(idx uint64, idxKey []byte) {
batch := new(leveldb.Batch)
batch.Delete(idxKey)
batch.Delete(getDataKey(idx))
dbStoreDeleteCounter.Inc(1)
s.entryCnt--
batch.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
s.db.Write(batch)

16
swarm/storage/localstore.go
View file

@ -18,6 +18,13 @@ package storage
import (
"encoding/binary"
"github.com/ethereum/go-ethereum/metrics"
)
//metrics variables
var (
dbStorePutCounter = metrics.NewCounter("storage.db.dbstore.put.count")
)
// LocalStore is a combination of inmemory db over a disk persisted db
@ -39,6 +46,14 @@ func NewLocalStore(hash SwarmHasher, params *StoreParams) (*LocalStore, error) {
}, nil
}
func (self *LocalStore) CacheCounter() uint64 {
return uint64(self.memStore.(*MemStore).Counter())
}
func (self *LocalStore) DbCounter() uint64 {
return self.DbStore.(*DbStore).Counter()
}
// LocalStore is itself a chunk store
// unsafe, in that the data is not integrity checked
func (self *LocalStore) Put(chunk *Chunk) {
@ -48,6 +63,7 @@ func (self *LocalStore) Put(chunk *Chunk) {
chunk.wg.Add(1)
}
go func() {
dbStorePutCounter.Inc(1)
self.DbStore.Put(chunk)
if chunk.wg != nil {
chunk.wg.Done()

14
swarm/storage/memstore.go
View file

@ -23,6 +23,13 @@ import (
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
)
//metrics variables
var (
memstorePutCounter = metrics.NewCounter("storage.db.memstore.put.count")
memstoreRemoveCounter = metrics.NewCounter("storage.db.memstore.rm.count")
)
const (
@ -130,6 +137,10 @@ func (s *MemStore) setCapacity(c uint) {
s.capacity = c
}
func (s *MemStore) Counter() uint {
return s.entryCnt
}
// entry (not its copy) is going to be in MemStore
func (s *MemStore) Put(entry *Chunk) {
if s.capacity == 0 {
@ -145,6 +156,8 @@ func (s *MemStore) Put(entry *Chunk) {
s.accessCnt++
memstorePutCounter.Inc(1)
node := s.memtree
bitpos := uint(0)
for node.entry == nil {
@ -289,6 +302,7 @@ func (s *MemStore) removeOldest() {
}
if node.entry.SData != nil {
memstoreRemoveCounter.Inc(1)
node.entry = nil
s.entryCnt--
}

37
swarm/swarm.go
View file

@ -34,6 +34,7 @@ import (
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/discover"
@ -44,6 +45,18 @@ import (
"github.com/ethereum/go-ethereum/swarm/fuse"
"github.com/ethereum/go-ethereum/swarm/network"
"github.com/ethereum/go-ethereum/swarm/storage"
gometrics "github.com/rcrowley/go-metrics"
)
var (
runTimer gometrics.Timer
startTime time.Time
metricsTimeout = 5 * time.Second
startCounter = metrics.NewCounter("stack,start")
stopCounter = metrics.NewCounter("stack,stop")
dbSizeGauge = metrics.NewGauge("storage.db.chunks.size")
cacheSizeGauge = metrics.NewGauge("storage.db.cache.size")
)
// the swarm stack
@ -262,6 +275,8 @@ Start is called when the stack is started
*/
// implements the node.Service interface
func (self *Swarm) Start(srv *p2p.Server) error {
runTimer = metrics.NewTimer("stack,uptime")
startTime = time.Now()
connectPeer := func(url string) error {
node, err := discover.ParseNode(url)
if err != nil {
@ -307,9 +322,29 @@ func (self *Swarm) Start(srv *p2p.Server) error {
}
}
go self.metricsLoop()
startCounter.Inc(1)
return nil
}
// metricsLoop periodically sends metrics about storage
func (self *Swarm) metricsLoop() {
ticker := time.NewTicker(metricsTimeout)
go func() {
for _ = range ticker.C {
self.sendMetrics()
}
}()
}
func (self *Swarm) sendMetrics() {
dbSizeGauge.Update(int64(self.lstore.DbCounter()))
cacheSizeGauge.Update(int64(self.lstore.CacheCounter()))
runTimer.UpdateSince(startTime)
}
// implements the node.Service interface
// stops all component services.
func (self *Swarm) Stop() error {
@ -324,6 +359,8 @@ func (self *Swarm) Stop() error {
self.lstore.DbStore.Close()
}
self.sfs.Stop()
stopCounter.Inc(1)
runTimer.UpdateSince(startTime)
return err
}