| 1 | package peerlog |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "sync/atomic" |
| 6 | "time" |
| 7 | |
| 8 | logging "github.com/ipfs/go-log/v2" |
| 9 | core "github.com/ipfs/kubo/core" |
| 10 | plugin "github.com/ipfs/kubo/plugin" |
| 11 | event "github.com/libp2p/go-libp2p/core/event" |
| 12 | network "github.com/libp2p/go-libp2p/core/network" |
| 13 | "github.com/libp2p/go-libp2p/core/peer" |
| 14 | "github.com/libp2p/go-libp2p/core/peerstore" |
| 15 | "go.uber.org/zap" |
| 16 | ) |
| 17 | |
| 18 | var log = logging.Logger("plugin/peerlog") |
| 19 | |
| 20 | type eventType int |
| 21 | |
| 22 | var ( |
| 23 | // size of the event queue buffer. |
| 24 | eventQueueSize = 64 * 1024 |
| 25 | // number of events to drop when busy. |
| 26 | busyDropAmount = eventQueueSize / 8 |
| 27 | ) |
| 28 | |
| 29 | const ( |
| 30 | eventConnect eventType = iota |
| 31 | eventIdentify |
| 32 | ) |
| 33 | |
| 34 | type plEvent struct { |
| 35 | kind eventType |
| 36 | peer peer.ID |
| 37 | } |
| 38 | |
| 39 | // Log all the PeerIDs. This is considered internal, unsupported, and may break at any point. |
| 40 | // |
| 41 | // Usage: |
| 42 | // |
| 43 | // GOLOG_FILE=~/peer.log GOLOG_LOG_FMT=json ipfs daemon |
| 44 | // |
| 45 | // Output: |
| 46 | // |
| 47 | // {"level":"info","ts":"2020-02-10T13:54:26.639Z","logger":"plugin/peerlog","caller":"peerlog/peerlog.go:51","msg":"connected","peer":"QmS2H72gdrekXJggGdE9SunXPntBqdkJdkXQJjuxcH8Cbt"} |
| 48 | // {"level":"info","ts":"2020-02-10T13:54:59.095Z","logger":"plugin/peerlog","caller":"peerlog/peerlog.go:56","msg":"identified","peer":"QmS2H72gdrekXJggGdE9SunXPntBqdkJdkXQJjuxcH8Cbt","agent":"go-ipfs/0.5.0/"} |
| 49 | type peerLogPlugin struct { |
| 50 | enabled bool |
| 51 | droppedCount uint64 |
| 52 | events chan plEvent |
| 53 | } |
| 54 | |
| 55 | var _ plugin.PluginDaemonInternal = (*peerLogPlugin)(nil) |
| 56 | |
| 57 | // Plugins is exported list of plugins that will be loaded. |
| 58 | var Plugins = []plugin.Plugin{ |
| 59 | &peerLogPlugin{}, |
| 60 | } |
| 61 | |
| 62 | // Name returns the plugin's name, satisfying the plugin.Plugin interface. |
| 63 | func (*peerLogPlugin) Name() string { |
| 64 | return "peerlog" |
| 65 | } |
| 66 | |
| 67 | // Version returns the plugin's version, satisfying the plugin.Plugin interface. |
| 68 | func (*peerLogPlugin) Version() string { |
| 69 | return "0.1.0" |
| 70 | } |
| 71 | |
| 72 | // extractEnabled extracts the "Enabled" field from the plugin config. |
| 73 | // Do not follow this as a precedent, this is only applicable to this plugin, |
| 74 | // since it is internal-only, unsupported functionality. |
| 75 | // For supported functionality, we should rework the plugin API to support this use case |
| 76 | // of including plugins that are disabled by default. |
| 77 | func extractEnabled(config any) bool { |
| 78 | // plugin is disabled by default, unless Enabled=true |
| 79 | if config == nil { |
| 80 | return false |
| 81 | } |
| 82 | mapIface, ok := config.(map[string]any) |
| 83 | if !ok { |
| 84 | return false |
| 85 | } |
| 86 | enabledIface, ok := mapIface["Enabled"] |
| 87 | if !ok || enabledIface == nil { |
| 88 | return false |
| 89 | } |
| 90 | enabled, ok := enabledIface.(bool) |
| 91 | if !ok { |
| 92 | return false |
| 93 | } |
| 94 | return enabled |
| 95 | } |
| 96 | |
| 97 | // Init initializes plugin. |
| 98 | func (pl *peerLogPlugin) Init(env *plugin.Environment) error { |
| 99 | pl.events = make(chan plEvent, eventQueueSize) |
| 100 | pl.enabled = extractEnabled(env.Config) |
| 101 | return nil |
| 102 | } |
| 103 | |
| 104 | func (pl *peerLogPlugin) collectEvents(node *core.IpfsNode) { |
| 105 | ctx := node.Context() |
| 106 | |
| 107 | busyCounter := 0 |
| 108 | dlog := log.Desugar() |
| 109 | for { |
| 110 | // Deal with dropped events. |
| 111 | dropped := atomic.SwapUint64(&pl.droppedCount, 0) |
| 112 | if dropped > 0 { |
| 113 | busyCounter++ |
| 114 | |
| 115 | // sleep a bit to give the system a chance to catch up with logging. |
| 116 | select { |
| 117 | case <-time.After(time.Duration(busyCounter) * time.Second): |
| 118 | case <-ctx.Done(): |
| 119 | return |
| 120 | } |
| 121 | |
| 122 | // drain 1/8th of the backlog backlog so we |
| 123 | // don't immediately run into this situation |
| 124 | // again. |
| 125 | loop: |
| 126 | for range busyDropAmount { |
| 127 | select { |
| 128 | case <-pl.events: |
| 129 | dropped++ |
| 130 | default: |
| 131 | break loop |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | // Add in any events we've dropped in the mean-time. |
| 136 | dropped += atomic.SwapUint64(&pl.droppedCount, 0) |
| 137 | |
| 138 | // Report that we've dropped events. |
| 139 | dlog.Error("dropped events", zap.Uint64("count", dropped)) |
| 140 | } else { |
| 141 | busyCounter = 0 |
| 142 | } |
| 143 | |
| 144 | var e plEvent |
| 145 | select { |
| 146 | case <-ctx.Done(): |
| 147 | return |
| 148 | case e = <-pl.events: |
| 149 | } |
| 150 | |
| 151 | peerID := zap.String("peer", e.peer.String()) |
| 152 | |
| 153 | switch e.kind { |
| 154 | case eventConnect: |
| 155 | dlog.Info("connected", peerID) |
| 156 | case eventIdentify: |
| 157 | agent, err := node.Peerstore.Get(e.peer, "AgentVersion") |
| 158 | switch err { |
| 159 | case nil: |
| 160 | case peerstore.ErrNotFound: |
| 161 | continue |
| 162 | default: |
| 163 | dlog.Error("failed to get agent version", zap.Error(err)) |
| 164 | continue |
| 165 | } |
| 166 | |
| 167 | agentS, ok := agent.(string) |
| 168 | if !ok { |
| 169 | continue |
| 170 | } |
| 171 | dlog.Info("identified", peerID, zap.String("agent", agentS)) |
| 172 | } |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | func (pl *peerLogPlugin) emit(evt eventType, p peer.ID) { |
| 177 | select { |
| 178 | case pl.events <- plEvent{kind: evt, peer: p}: |
| 179 | default: |
| 180 | atomic.AddUint64(&pl.droppedCount, 1) |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | func (pl *peerLogPlugin) Start(node *core.IpfsNode) error { |
| 185 | if !pl.enabled { |
| 186 | return nil |
| 187 | } |
| 188 | |
| 189 | // Ensure logs from this plugin get printed regardless of global GOLOG_LOG_LEVEL value |
| 190 | if err := logging.SetLogLevel("plugin/peerlog", "info"); err != nil { |
| 191 | return fmt.Errorf("failed to set log level: %w", err) |
| 192 | } |
| 193 | |
| 194 | sub, err := node.PeerHost.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted)) |
| 195 | if err != nil { |
| 196 | return fmt.Errorf("failed to subscribe to identify notifications") |
| 197 | } |
| 198 | |
| 199 | var notifee network.NotifyBundle |
| 200 | notifee.ConnectedF = func(net network.Network, conn network.Conn) { |
| 201 | pl.emit(eventConnect, conn.RemotePeer()) |
| 202 | } |
| 203 | node.PeerHost.Network().Notify(¬ifee) |
| 204 | |
| 205 | go func() { |
| 206 | defer sub.Close() |
| 207 | for e := range sub.Out() { |
| 208 | switch e := e.(type) { |
| 209 | case event.EvtPeerIdentificationCompleted: |
| 210 | pl.emit(eventIdentify, e.Peer) |
| 211 | } |
| 212 | } |
| 213 | }() |
| 214 | |
| 215 | go pl.collectEvents(node) |
| 216 | |
| 217 | return nil |
| 218 | } |
| 219 | |
| 220 | func (*peerLogPlugin) Close() error { |
| 221 | return nil |
| 222 | } |