Merge pull request #570 from jbenet/feat/eventbegin

Feat/eventbegin

exchange/bitswap/bitswap.go

@@ -120,12 +120,11 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err
 	ctx, cancelFunc := context.WithCancel(parent)
 
 	ctx = eventlog.ContextWithLoggable(ctx, eventlog.Uuid("GetBlockRequest"))
-	log.Event(ctx, "GetBlockRequestBegin", &k)
+	defer log.EventBegin(ctx, "GetBlockRequest", &k).Done()
 	log.Debugf("GetBlockRequestBegin")
 
 	defer func() {
 		cancelFunc()
-		log.Event(ctx, "GetBlockRequestEnd", &k)
 		log.Debugf("GetBlockRequestEnd")
 	}()
 

p2p/crypto/secio/protocol.go

@@ -81,7 +81,8 @@ func (s *secureSession) handshake(ctx context.Context, insecure io.ReadWriter) e
 	}
 
 	log.Debugf("handshake: %s <--start--> %s", s.localPeer, s.remotePeer)
-	log.Event(ctx, "secureHandshakeStart", s.localPeer)
+	defer log.EventBegin(ctx, "secureHandshake", s.localPeer).Done()
+
 	s.local.permanentPubKey = s.localKey.GetPublic()
 	myPubKeyBytes, err := s.local.permanentPubKey.Bytes()
 	if err != nil {

routing/dht/dht.go

@@ -196,6 +196,7 @@ func (dht *IpfsDHT) getValueOrPeers(ctx context.Context, p peer.ID,
 // getValueSingle simply performs the get value RPC with the given parameters
 func (dht *IpfsDHT) getValueSingle(ctx context.Context, p peer.ID,
 	key u.Key) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "getValueSingle", p, &key).Done()
 
 	pmes := pb.NewMessage(pb.Message_GET_VALUE, string(key), 0)
 	return dht.sendRequest(ctx, p, pmes)
@@ -265,11 +266,15 @@ func (dht *IpfsDHT) FindLocal(id peer.ID) peer.PeerInfo {
 
 // findPeerSingle asks peer 'p' if they know where the peer with id 'id' is
 func (dht *IpfsDHT) findPeerSingle(ctx context.Context, p peer.ID, id peer.ID) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "findPeerSingle", p, id).Done()
+
 	pmes := pb.NewMessage(pb.Message_FIND_NODE, string(id), 0)
 	return dht.sendRequest(ctx, p, pmes)
 }
 
 func (dht *IpfsDHT) findProvidersSingle(ctx context.Context, p peer.ID, key u.Key) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "findProvidersSingle", p, &key).Done()
+
 	pmes := pb.NewMessage(pb.Message_GET_PROVIDERS, string(key), 0)
 	return dht.sendRequest(ctx, p, pmes)
 }

routing/dht/handlers.go

@@ -39,6 +39,7 @@ func (dht *IpfsDHT) handlerForMsgType(t pb.Message_MessageType) dhtHandler {
 }
 
 func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "handleGetValue", p).Done()
 	log.Debugf("%s handleGetValue for key: %s", dht.self, pmes.GetKey())
 
 	// setup response
@@ -114,6 +115,7 @@ func (dht *IpfsDHT) handleGetValue(ctx context.Context, p peer.ID, pmes *pb.Mess
 
 // Store a value in this peer local storage
 func (dht *IpfsDHT) handlePutValue(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "handlePutValue", p).Done()
 	dskey := u.Key(pmes.GetKey()).DsKey()
 
 	if err := dht.verifyRecordLocally(pmes.GetRecord()); err != nil {
@@ -137,6 +139,7 @@ func (dht *IpfsDHT) handlePing(_ context.Context, p peer.ID, pmes *pb.Message) (
 }
 
 func (dht *IpfsDHT) handleFindPeer(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "handleFindPeer", p).Done()
 	resp := pb.NewMessage(pmes.GetType(), "", pmes.GetClusterLevel())
 	var closest []peer.ID
 
@@ -166,6 +169,7 @@ func (dht *IpfsDHT) handleFindPeer(ctx context.Context, p peer.ID, pmes *pb.Mess
 }
 
 func (dht *IpfsDHT) handleGetProviders(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "handleGetProviders", p).Done()
 	resp := pb.NewMessage(pmes.GetType(), pmes.GetKey(), pmes.GetClusterLevel())
 	key := u.Key(pmes.GetKey())
 
@@ -211,6 +215,7 @@ type providerInfo struct {
 }
 
 func (dht *IpfsDHT) handleAddProvider(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
+	defer log.EventBegin(ctx, "handleAddProvider", p).Done()
 	key := u.Key(pmes.GetKey())
 
 	log.Debugf("%s adding %s as a provider for '%s'\n", dht.self, p, key)

routing/dht/routing.go

@@ -122,10 +122,8 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key u.Key) ([]byte, error) {
 // Provide makes this node announce that it can provide a value for the given key
 func (dht *IpfsDHT) Provide(ctx context.Context, key u.Key) error {
 	log := dht.log().Prefix("Provide(%s)", key)
-	log.Debugf("start", key)
-	log.Event(ctx, "provideBegin", &key)
-	defer log.Debugf("end", key)
-	defer log.Event(ctx, "provideEnd", &key)
+
+	defer log.EventBegin(ctx, "provide", &key).Done()
 
 	// add self locally
 	dht.providers.AddProvider(key, dht.self)
@@ -163,6 +161,7 @@ func (dht *IpfsDHT) FindProviders(ctx context.Context, key u.Key) ([]peer.PeerIn
 // Kademlia 'node lookup' operation. Returns a channel of the K closest peers
 // to the given key
 func (dht *IpfsDHT) getClosestPeers(ctx context.Context, key u.Key) (<-chan peer.ID, error) {
+	e := log.EventBegin(ctx, "getClosestPeers", &key)
 	tablepeers := dht.routingTable.NearestPeers(kb.ConvertKey(key), AlphaValue)
 	if len(tablepeers) == 0 {
 		return nil, errors.Wrap(kb.ErrLookupFailure)
@@ -204,6 +203,7 @@ func (dht *IpfsDHT) getClosestPeers(ctx context.Context, key u.Key) (<-chan peer
 
 	go func() {
 		defer close(out)
+		defer e.Done()
 		// run it!
 		_, err := query.Run(ctx, tablepeers)
 		if err != nil {
@@ -242,10 +242,8 @@ func (dht *IpfsDHT) FindProvidersAsync(ctx context.Context, key u.Key, count int
 func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, count int, peerOut chan peer.PeerInfo) {
 	log := dht.log().Prefix("FindProviders(%s)", key)
 
+	defer log.EventBegin(ctx, "findProvidersAsync", &key).Done()
 	defer close(peerOut)
-	defer log.Event(ctx, "findProviders end", &key)
-	log.Debug("begin")
-	defer log.Debug("begin")
 
 	ps := pset.NewLimited(count)
 	provs := dht.providers.GetProviders(ctx, key)
@@ -314,6 +312,7 @@ func (dht *IpfsDHT) findProvidersAsyncRoutine(ctx context.Context, key u.Key, co
 
 // FindPeer searches for a peer with given ID.
 func (dht *IpfsDHT) FindPeer(ctx context.Context, id peer.ID) (peer.PeerInfo, error) {
+	defer log.EventBegin(ctx, "FindPeer", id).Done()
 
 	// Check if were already connected to them
 	if pi := dht.FindLocal(id); pi.ID != "" {