diff --git a/cmd/p2p/sensor/sensor.go b/cmd/p2p/sensor/sensor.go index cacd1e029..759b3c1f1 100644 --- a/cmd/p2p/sensor/sensor.go +++ b/cmd/p2p/sensor/sensor.go @@ -42,6 +42,7 @@ type ( TrustedNodesFile string ProjectID string DatabaseID string + ClickHouseDSN string SensorID string MaxPeers int MaxDatabaseConcurrency int @@ -483,6 +484,21 @@ func newDatabase(ctx context.Context) (database.Database, error) { ShouldWritePeers: inputSensorParams.ShouldWritePeers, TTL: inputSensorParams.TTL, }), nil + case "clickhouse": + return database.NewClickHouse(ctx, database.ClickHouseOptions{ + DSN: inputSensorParams.ClickHouseDSN, + SensorID: inputSensorParams.SensorID, + ChainID: inputSensorParams.NetworkID, + MaxConcurrency: inputSensorParams.MaxDatabaseConcurrency, + ShouldWriteBlocks: inputSensorParams.ShouldWriteBlocks, + ShouldWriteBlockEvents: inputSensorParams.ShouldWriteBlockEvents, + ShouldWriteFirstBlockEvent: inputSensorParams.ShouldWriteFirstBlockEvent, + ShouldWriteTransactions: inputSensorParams.ShouldWriteTransactions, + ShouldWriteTransactionEvents: inputSensorParams.ShouldWriteTransactionEvents, + ShouldWriteFirstTransactionEvent: inputSensorParams.ShouldWriteFirstTransactionEvent, + ShouldWritePeers: inputSensorParams.ShouldWritePeers, + TTL: inputSensorParams.TTL, + }), nil case "json": return database.NewJSONDatabase(database.JSONDatabaseOptions{ SensorID: inputSensorParams.SensorID, @@ -508,6 +524,8 @@ func init() { flag.MarkFlagsRequired(SensorCmd, "network-id") f.StringVarP(&inputSensorParams.ProjectID, "project-id", "p", "", "GCP project ID") f.StringVarP(&inputSensorParams.DatabaseID, "database-id", "d", "", "datastore database ID") + f.StringVar(&inputSensorParams.ClickHouseDSN, "clickhouse-dsn", "", + "ClickHouse DSN, e.g. clickhouse://user:pass@host:9000/sensor (used with --database=clickhouse)") f.StringVarP(&inputSensorParams.SensorID, "sensor-id", "s", "", "sensor ID when writing block/tx events") flag.MarkFlagsRequired(SensorCmd, "sensor-id") f.IntVarP(&inputSensorParams.MaxPeers, "max-peers", "m", 2000, "maximum number of peers to connect to") @@ -564,6 +582,7 @@ will result in less chance of missing data but can significantly increase memory f.StringVar(&inputSensorParams.Database, "database", "none", `which database to persist data to, options are: - datastore (GCP Datastore) + - clickhouse (ClickHouse, see --clickhouse-dsn) - json (output to stdout) - none (no persistence)`) f.BoolVar(&inputSensorParams.NoDiscovery, "no-discovery", false, "disable P2P peer discovery") diff --git a/doc/polycli_p2p_sensor.md b/doc/polycli_p2p_sensor.md index 5cdb07148..320835bc1 100644 --- a/doc/polycli_p2p_sensor.md +++ b/doc/polycli_p2p_sensor.md @@ -152,8 +152,10 @@ polycli p2p sensor amoy-nodes.json \ --broadcast-txs broadcast full transactions to peers --broadcast-workers int number of concurrent broadcast workers (default 4) --cache-only-validated-blocks only cache and serve blocks signed by a known validator (unknown-signer blocks are still recorded to the database); has no effect without --validate-block-signer (default true) + --clickhouse-dsn string ClickHouse DSN, e.g. clickhouse://user:pass@host:9000/sensor (used with --database=clickhouse) --database string which database to persist data to, options are: - datastore (GCP Datastore) + - clickhouse (ClickHouse, see --clickhouse-dsn) - json (output to stdout) - none (no persistence) (default "none") -d, --database-id string datastore database ID diff --git a/go.mod b/go.mod index 870374a86..2029f7933 100644 --- a/go.mod +++ b/go.mod @@ -38,15 +38,16 @@ require ( require github.com/alecthomas/participle/v2 v2.1.4 require ( + github.com/ClickHouse/ch-go v0.73.0 // indirect github.com/ProjectZKM/Ziren/crates/go-runtime/zkvm_runtime v0.0.0-20251213223233-751f36331c62 // indirect + github.com/andybalholm/brotli v1.2.1 // indirect github.com/chromedp/sysutil v1.1.0 // indirect github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect - github.com/containerd/log v0.1.0 // indirect github.com/crate-crypto/go-eth-kzg v1.5.0 // indirect github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect github.com/distribution/reference v0.6.0 // indirect - github.com/docker/go-connections v0.4.0 // indirect + github.com/docker/go-connections v0.7.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/emicklei/dot v1.6.2 // indirect github.com/ethereum/c-kzg-4844/v2 v2.1.6 // indirect @@ -54,6 +55,8 @@ require ( github.com/ferranbt/fastssz v0.1.4 // indirect github.com/fjl/jsonw v0.1.0 // indirect github.com/gdamore/encoding v1.0.1 // indirect + github.com/go-faster/city v1.0.1 // indirect + github.com/go-faster/errors v0.7.1 // indirect github.com/go-json-experiment/json v0.0.0-20260214004413-d219187c3433 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/gobwas/httphead v0.1.0 // indirect @@ -66,11 +69,15 @@ require ( github.com/morikuni/aec v1.0.0 // indirect github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/paulmach/orb v0.13.0 // indirect + github.com/pierrec/lz4/v4 v4.1.27 // indirect github.com/pion/dtls/v3 v3.1.2 // indirect github.com/pion/logging v0.2.4 // indirect github.com/pion/stun/v3 v3.1.2 // indirect github.com/pion/transport/v4 v4.0.1 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect + github.com/segmentio/asm v1.2.1 // indirect + github.com/shopspring/decimal v1.4.0 // indirect github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect @@ -119,7 +126,7 @@ require ( github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/ipfs/go-cid v0.4.1 // indirect github.com/jackpal/go-nat-pmp v1.0.2 // indirect - github.com/klauspost/compress v1.18.0 // indirect + github.com/klauspost/compress v1.18.6 // indirect github.com/klauspost/cpuid/v2 v2.2.8 // indirect github.com/kr/pretty v0.3.1 // indirect github.com/kr/text v0.2.0 // indirect @@ -148,8 +155,8 @@ require ( github.com/shirou/gopsutil v3.21.11+incompatible // indirect github.com/spaolacci/murmur3 v1.1.0 // indirect github.com/supranational/blst v0.3.16 // indirect - github.com/tklauser/go-sysconf v0.3.14 // indirect - github.com/tklauser/numcpus v0.9.0 // indirect + github.com/tklauser/go-sysconf v0.4.0 // indirect + github.com/tklauser/numcpus v0.12.0 // indirect github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect go.opencensus.io v0.24.0 // indirect @@ -169,6 +176,7 @@ require ( require ( cloud.google.com/go/kms v1.31.0 github.com/0xPolygon/cdk-contracts-tooling v0.0.1 + github.com/ClickHouse/clickhouse-go/v2 v2.47.0 github.com/btcsuite/btcd/btcutil v1.2.0 github.com/chromedp/cdproto v0.0.0-20260321001828-e3e3800016bc github.com/chromedp/chromedp v0.15.1 @@ -221,8 +229,8 @@ require ( github.com/yusufpapurcu/wmi v1.2.4 // indirect go.etcd.io/bbolt v1.4.0-alpha.0.0.20240404170359-43604f3112c5 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 // indirect - go.opentelemetry.io/otel v1.43.0 // indirect - go.opentelemetry.io/otel/metric v1.43.0 // indirect - go.opentelemetry.io/otel/trace v1.43.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect ) diff --git a/go.sum b/go.sum index 16c37e3a8..8b4790440 100644 --- a/go.sum +++ b/go.sum @@ -19,9 +19,13 @@ github.com/0xPolygon/cdk-contracts-tooling v0.0.1 h1:2HH8KpO1CZRl1zHfn0IYwJhPA7l github.com/0xPolygon/cdk-contracts-tooling v0.0.1/go.mod h1:mFlcEjsm2YBBsu8atHJ3zyVnwM+Z/fMXpVmIJge+WVU= github.com/0xPolygon/cdk-rpc v0.0.0-20250213125803-179882ad6229 h1:6YhqNQVcXkoxqs5zQVg1bREuoeKvwpffpfoL8QQT+u4= github.com/0xPolygon/cdk-rpc v0.0.0-20250213125803-179882ad6229/go.mod h1:2scWqMMufrQXu7TikDgQ3BsyaKoX8qP26D6E262vSOg= -github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0= -github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg= +github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/ClickHouse/ch-go v0.73.0 h1:jsHiGRbQ3sz+gekvDFJF29LWDo5dzbJm5s1h8TWVP2M= +github.com/ClickHouse/ch-go v0.73.0/go.mod h1:wkFIxrqlXeRJ9cn3r5Fz5Qen9jl5aTMPuGZeuJpANNY= +github.com/ClickHouse/clickhouse-go/v2 v2.47.0 h1:ZDAzrnKSOPTIsm4tdUNfrii2yc8dk4SVRLC77BR7Z5Q= +github.com/ClickHouse/clickhouse-go/v2 v2.47.0/go.mod h1:sPj7C7UYQ2MWHcfX+4eGN6nwnCqwUKfgO6PcwKpd6K8= github.com/DataDog/zstd v1.5.6 h1:LbEglqepa/ipmmQJUDnSsfvA8e8IStVcGaFWDuxvGOY= github.com/DataDog/zstd v1.5.6/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw= github.com/FactomProject/basen v0.0.0-20150613233007-fe3947df716e h1:ahyvB3q25YnZWly5Gq1ekg6jcmWaGj/vG/MhF4aisoc= @@ -43,6 +47,8 @@ github.com/alecthomas/repr v0.4.0 h1:GhI2A8MACjfegCPVq9f1FLvIBS+DrQ2KQBFZP1iFzXc github.com/alecthomas/repr v0.4.0/go.mod h1:Fr0507jx4eOXV7AlPV6AVZLYrLIuIeSOWtW57eE/O/4= github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156 h1:eMwmnE/GDgah4HI848JfFxHt+iPb26b4zyfspmqY0/8= github.com/allegro/bigcache v1.2.1-0.20190218064605-e24eb225f156/go.mod h1:Cb/ax3seSYIx7SuZdm2G2xzfwmv3TPSk2ucNfQESPXM= +github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro= +github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bits-and-blooms/bitset v1.24.4 h1:95H15Og1clikBrKr/DuzMXkQzECs1M6hhoGXLwLQOZE= @@ -155,8 +161,8 @@ github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5Qvfr github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM= github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= -github.com/docker/go-connections v0.4.0 h1:El9xVISelRB7BuFusrZozjnkIM5YnzCViNKohAFqRJQ= -github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec= +github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c= +github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk= @@ -212,6 +218,10 @@ github.com/go-echarts/go-echarts/v2 v2.7.2 h1:lhypL1CekgqaLHM5V7fBPfaYGfimJ9dGyl github.com/go-echarts/go-echarts/v2 v2.7.2/go.mod h1:Z+spPygZRIEyqod69r0WMnkN5RV3MwhYDtw601w3G8w= github.com/go-errors/errors v1.5.1 h1:ZwEMSLRCapFLflTpT7NKaAc7ukJ8ZPEjzlxt8rPN8bk= github.com/go-errors/errors v1.5.1/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw= +github.com/go-faster/city v1.0.1/go.mod h1:jKcUJId49qdW3L1qKHH/3wPeUstCVpVSXTM6vO3VcTw= +github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg= +github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo= github.com/go-json-experiment/json v0.0.0-20260214004413-d219187c3433 h1:vymEbVwYFP/L05h5TKQxvkXoKxNvTpjxYKdF1Nlwuao= github.com/go-json-experiment/json v0.0.0-20260214004413-d219187c3433/go.mod h1:tphK2c80bpPhMOI4v6bIc2xWywPfbqi1Z06+RcrMkDg= github.com/go-kit/log v0.2.1 h1:MRVx0/zhvdseW+Gza6N9rVzU/IVzaeE1SFI4raAhmBU= @@ -341,8 +351,8 @@ github.com/jrick/logrotate v1.0.0/go.mod h1:LNinyqDIJnpAur+b8yyulnQw/wDuN1+BYKlT github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/kkdai/bstream v0.0.0-20161212061736-f391b8402d23/go.mod h1:J+Gs4SYgM6CZQHDETBtE9HaSEkGmuNXF86RwHhHUvq4= -github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao= +github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.2.8 h1:+StwCXwm9PdpiEkPyzBXIy+M9KUb4ODm0Zarf1kS5BM= github.com/klauspost/cpuid/v2 v2.2.8/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= @@ -389,8 +399,8 @@ github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w github.com/moby/sys/atomicwriter v0.1.0/go.mod h1:Ul8oqv2ZMNHOceF643P6FKPXeCmYtlQMvpizfsSoaWs= github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU= github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko= -github.com/moby/term v0.0.0-20221205130635-1aeaba878587 h1:HfkjXDfhgVaN5rmueG8cL8KKeFNecRCXFhaJ2qZ5SKA= -github.com/moby/term v0.0.0-20221205130635-1aeaba878587/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y= +github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= +github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= github.com/montanaflynn/stats v0.10.0 h1:8GQhct8Qup299/6gFmW3K3xRvPrdpmdjVCWRsZ5bnes= github.com/montanaflynn/stats v0.10.0/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow= github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= @@ -443,12 +453,16 @@ github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsq github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde h1:x0TT0RDC7UhAVbbWWBzr41ElhJx5tXPWkIHA2HWPRuw= github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0= +github.com/paulmach/orb v0.13.0 h1:r7n7mQGGF+cj/CbcivEj9J3HGK+XR+yXnvzRdq9saIw= +github.com/paulmach/orb v0.13.0/go.mod h1:6scRWINywA2Jf05dcjOfLfxrUIMECvTSG2MVbRLxu/k= github.com/pelletier/go-toml/v2 v2.4.3 h1:GTRvJQutkOSftxIFD5xw9aepkYNuPWmVJpffdDPYVpY= github.com/pelletier/go-toml/v2 v2.4.3/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7 h1:oYW+YCJ1pachXTQmzR3rNLYGGz4g/UgFcjb28p/viDM= github.com/peterh/liner v1.1.1-0.20190123174540-a2c9a5303de7/go.mod h1:CRroGNssyjTd/qIG2FyxByd2S8JEAZXBl4qUrZf8GS0= github.com/petermattis/goid v0.0.0-20240813172612-4fcff4a6cae7 h1:Dx7Ovyv/SFnMFw3fD4oEoeorXc6saIiQ23LrGLth0Gw= github.com/petermattis/goid v0.0.0-20240813172612-4fcff4a6cae7/go.mod h1:pxMtw7cyUw6B2bRH0ZBANSPg+AoSud1I1iyJHI69jH4= +github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= +github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4= github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= github.com/pion/dtls/v3 v3.1.2 h1:gqEdOUXLtCGW+afsBLO0LtDD8GnuBBjEy6HRtyofZTc= @@ -500,10 +514,14 @@ github.com/sasha-s/go-deadlock v0.3.5 h1:tNCOEEDG6tBqrNDOX35j/7hL5FcFViG6awUGROb github.com/sasha-s/go-deadlock v0.3.5/go.mod h1:bugP6EGbdGYObIlx7pUZtWqlvo8k9H6vCBBsiChJQ5U= github.com/schollz/progressbar/v3 v3.19.1 h1:iv8BgwOvdML/S3p84uBpy/IMigv4U9594vPZYa2EdrU= github.com/schollz/progressbar/v3 v3.19.1/go.mod h1:LFL7jqimKxfhero4K1eCkUr/6R39AgQeiPCJtlTWIW8= +github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= +github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI= github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= -github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= -github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= +github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 h1:+jumHNA0Wrelhe64i8F6HNlS8pkoyMv5sreGx2Ry5Rw= github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8/go.mod h1:3n1Cwaq1E1/1lhQhtRK2ts/ZwZEhjcQeJQ1RuC6Q/8U= github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= @@ -538,10 +556,10 @@ github.com/supranational/blst v0.3.16 h1:bTDadT+3fK497EvLdWRQEjiGnUtzJ7jjIUMF0jq github.com/supranational/blst v0.3.16/go.mod h1:jZJtfjgudtNl4en1tzwPIV3KjUnQUvG3/j+w+fVonLw= github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d h1:vfofYNRScrDdvS342BElfbETmL1Aiz3i2t0zfRj16Hs= github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d/go.mod h1:RRCYJbIwD5jmqPI9XoAFR0OcDxqUctll6zUj/+B4S48= -github.com/tklauser/go-sysconf v0.3.14 h1:g5vzr9iPFFz24v2KZXs/pvpvh8/V9Fw6vQK5ZZb78yU= -github.com/tklauser/go-sysconf v0.3.14/go.mod h1:1ym4lWMLUOhuBOPGtRcJm7tEGX4SCYNEEEtghGG/8uY= -github.com/tklauser/numcpus v0.9.0 h1:lmyCHtANi8aRUgkckBgoDk1nHCux3n2cgkJLXdQGPDo= -github.com/tklauser/numcpus v0.9.0/go.mod h1:SN6Nq1O3VychhC1npsWostA+oW+VOQTxZrS604NSRyI= +github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU= +github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI= +github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4= +github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg= github.com/tyler-smith/go-bip32 v1.0.0 h1:sDR9juArbUgX+bO/iblgZnMPeWY1KZMUC2AFUJdv5KE= github.com/tyler-smith/go-bip32 v1.0.0/go.mod h1:onot+eHknzV4BVPwrzqY5OoVpyCvnwD7lMawL5aQupE= github.com/tyler-smith/go-bip39 v1.1.0 h1:5eUemwrMargf3BSLRRCalXT93Ns6pQJIjYQN2nyfOP8= @@ -559,6 +577,8 @@ github.com/xeipuuv/gojsonschema v1.2.0 h1:LhYJRs+L4fBtjZUfuSZIKGeVu0QRy8e5Xi7D17 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1 h1:gEOO8jv9F4OT7lGCjxCBTO/36wtF6j2nSip77qHd4x4= github.com/xrash/smetrics v0.0.0-20240521201337-686a1a2994c1/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM= +github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= @@ -574,22 +594,22 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 h1:yI1/OhfEPy7J9eoa6Sj051C7n5dvpj0QX8g4sRchg04= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0/go.mod h1:NoUCKYWK+3ecatC4HjkRktREheMeEtrXoQxrqYFeHSc= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 h1:OyrsyzuttWTSur2qN/Lm0m2a8yqyIjUVBZcxFPuXq2o= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0/go.mod h1:C2NGBr+kAB4bk3xtMXfZ94gqFDtg/GkI7e9zqGh5Beg= -go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= -go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 h1:8tvICD4vSTOOsNrsI4Ljf6C+6UKvpTEH5XY3JMoyPoo= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0/go.mod h1:z9+yiacE0IHRqM4qFfkbt/JYlmYXgss8GY/jXoNuPJI= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.41.0 h1:ao6Oe+wSebTlQ1OEht7jlYTzQKE+pnx/iNywFvTbuuI= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.41.0/go.mod h1:u3T6vz0gh/NVzgDgiwkgLxpsSF6PaPmo2il0apGJbls= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.41.0 h1:inYW9ZhgqiDqh6BioM7DVHHzEGVq76Db5897WLGZ5Go= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.41.0/go.mod h1:Izur+Wt8gClgMJqO/cZ8wdeeMryJ/xxiOVgFSSfpDTY= -go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= -go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= -go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= -go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= -go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= -go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= -go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= -go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= diff --git a/p2p/database/clickhouse.go b/p2p/database/clickhouse.go new file mode 100644 index 000000000..5d83c6702 --- /dev/null +++ b/p2p/database/clickhouse.go @@ -0,0 +1,538 @@ +package database + +import ( + "context" + "fmt" + "math/big" + "sync/atomic" + "time" + + "github.com/0xPolygon/polygon-cli/util" + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/eth/protocols/eth" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/enode" + "github.com/rs/zerolog/log" +) + +// Default batching parameters. ClickHouse strongly prefers large, infrequent +// inserts over many small ones, so every Write* call only enqueues a row into +// an in-memory buffer; a background goroutine flushes it in batches. This keeps +// the sensor hot path non-blocking and turns the write pattern into the bulk +// append ClickHouse is built for. Buffers are fixed-size and drop-on-full so a +// slow/unavailable database can never exhaust memory or stall the sensor. +const ( + chFlushInterval = 1 * time.Second + chFlushTimeout = 30 * time.Second + chBlockBatch = 5000 + chBlockEventBatch = 50000 + chTxBatch = 20000 + chTxEventBatch = 50000 + chPeerBatch = 2000 +) + +// ClickHouse implements the Database interface backed by a ClickHouse cluster. +// See clickhouse_schema.sql for the table definitions this writer targets. +type ClickHouse struct { + conn driver.Conn + sensorID string + chainID *big.Int + maxConcurrency int + shouldWriteBlocks bool + shouldWriteBlockEvents bool + shouldWriteFirstBlockEvent bool + shouldWriteTransactions bool + shouldWriteTransactionEvents bool + shouldWriteFirstTransactionEvent bool + shouldWritePeers bool + + blocks *rowBatcher[chBlock] + blockEvt *rowBatcher[chEvent] + txs *rowBatcher[chTx] + txEvt *rowBatcher[chEvent] + peers *rowBatcher[chPeer] +} + +// ClickHouseOptions is used when creating a NewClickHouse. +type ClickHouseOptions struct { + DSN string + SensorID string + ChainID uint64 + MaxConcurrency int + ShouldWriteBlocks bool + ShouldWriteBlockEvents bool + ShouldWriteFirstBlockEvent bool + ShouldWriteTransactions bool + ShouldWriteTransactionEvents bool + ShouldWriteFirstTransactionEvent bool + ShouldWritePeers bool + TTL time.Duration +} + +// NewClickHouse connects to ClickHouse, verifies connectivity, and starts the +// background batch flushers. The provided context governs the lifetime of the +// flusher goroutines; when it is cancelled they flush any buffered rows and +// exit. If the connection cannot be established the returned Database no-ops all +// writes (mirroring the Datastore backend) so the sensor keeps running. +func NewClickHouse(ctx context.Context, opts ClickHouseOptions) Database { + c := &ClickHouse{ + sensorID: opts.SensorID, + chainID: new(big.Int).SetUint64(opts.ChainID), + maxConcurrency: opts.MaxConcurrency, + shouldWriteBlocks: opts.ShouldWriteBlocks, + shouldWriteBlockEvents: opts.ShouldWriteBlockEvents, + shouldWriteFirstBlockEvent: opts.ShouldWriteFirstBlockEvent, + shouldWriteTransactions: opts.ShouldWriteTransactions, + shouldWriteTransactionEvents: opts.ShouldWriteTransactionEvents, + shouldWriteFirstTransactionEvent: opts.ShouldWriteFirstTransactionEvent, + shouldWritePeers: opts.ShouldWritePeers, + } + + conn, err := connectClickHouse(ctx, opts.DSN) + if err != nil { + log.Error().Err(err).Msg("Could not initialize ClickHouse connection") + return c + } + c.conn = conn + + c.startBatchers(ctx) + + return c +} + +// connectClickHouse parses the DSN, opens a connection, and verifies +// connectivity with a ping. +func connectClickHouse(ctx context.Context, dsn string) (driver.Conn, error) { + chOpts, err := clickhouse.ParseDSN(dsn) + if err != nil { + return nil, fmt.Errorf("could not parse ClickHouse DSN: %w", err) + } + + conn, err := clickhouse.Open(chOpts) + if err != nil { + return nil, fmt.Errorf("could not connect to ClickHouse: %w", err) + } + + if err := conn.Ping(ctx); err != nil { + return nil, fmt.Errorf("could not ping ClickHouse: %w", err) + } + return conn, nil +} + +// startBatchers creates the background row batchers, one per target table. Each +// batcher's append closure only needs to map a row to its column values; the +// surrounding batch/flush/error handling lives in newInsertBatcher. +func (c *ClickHouse) startBatchers(ctx context.Context) { + c.blocks = newInsertBatcher(ctx, c, "blocks", chBlockBatch, + "INSERT INTO blocks (hash, number, parent_hash, block_time, coinbase, signer, difficulty, total_difficulty, gas_used, gas_limit, base_fee, tx_count, uncle_count, uncle_hash, state_root, tx_root, receipt_root, logs_bloom, extra_data, mix_digest, nonce, sensor_id, ingested_at)", + func(b driver.Batch, r chBlock) error { + return b.Append(r.hash, r.number, r.parentHash, r.blockTime, r.coinbase, r.signer, r.difficulty, r.totalDifficulty, r.gasUsed, r.gasLimit, r.baseFee, r.txCount, r.uncleCount, r.uncleHash, r.stateRoot, r.txRoot, r.receiptRoot, r.logsBloom, r.extraData, r.mixDigest, r.nonce, c.sensorID, r.ingestedAt) + }) + c.blockEvt = newInsertBatcher(ctx, c, "block_events", chBlockEventBatch, + "INSERT INTO block_events (block_hash, sensor_id, peer_id, seen_at)", + func(b driver.Batch, r chEvent) error { + return b.Append(r.hash, c.sensorID, r.peerID, r.seenAt) + }) + c.txs = newInsertBatcher(ctx, c, "transactions", chTxBatch, + "INSERT INTO transactions (hash, from_address, to_address, value, gas, gas_price, gas_fee_cap, gas_tip_cap, nonce, tx_type, first_seen, sensor_id, ingested_at)", + func(b driver.Batch, r chTx) error { + return b.Append(r.hash, r.from, r.to, r.value, r.gas, r.gasPrice, r.gasFeeCap, r.gasTipCap, r.nonce, r.txType, r.firstSeen, c.sensorID, r.ingestedAt) + }) + c.txEvt = newInsertBatcher(ctx, c, "transaction_events", chTxEventBatch, + "INSERT INTO transaction_events (tx_hash, sensor_id, peer_id, seen_at)", + func(b driver.Batch, r chEvent) error { + return b.Append(r.hash, c.sensorID, r.peerID, r.seenAt) + }) + c.peers = newInsertBatcher(ctx, c, "peers", chPeerBatch, + "INSERT INTO peers (peer_id, name, url, caps, last_seen_by, time_last_seen)", + func(b driver.Batch, r chPeer) error { + return b.Append(r.peerID, r.name, r.url, r.caps, c.sensorID, r.timeLastSeen) + }) +} + +// newInsertBatcher wraps newRowBatcher with the common flush behaviour: prepare +// the INSERT, append each row via appendRow, and send. Only appendRow varies +// per table. +func newInsertBatcher[T any](ctx context.Context, c *ClickHouse, name string, maxRows int, query string, appendRow func(driver.Batch, T) error) *rowBatcher[T] { + return newRowBatcher(ctx, name, maxRows, func(fctx context.Context, rows []T) error { + return c.flush(fctx, query, func(b driver.Batch) error { + for _, r := range rows { + if err := appendRow(b, r); err != nil { + return err + } + } + return nil + }) + }) +} + +// flush prepares a batch for the given INSERT, appends every row via the +// provided callback, and sends it. It uses a detached, time-bounded context so +// a flush triggered during shutdown (parent context already cancelled) still +// completes. +func (c *ClickHouse) flush(_ context.Context, query string, append func(driver.Batch) error) error { + fctx, cancel := context.WithTimeout(context.Background(), chFlushTimeout) + defer cancel() + + b, err := c.conn.PrepareBatch(fctx, query) + if err != nil { + return err + } + if err := append(b); err != nil { + return err + } + return b.Send() +} + +// --- row types ------------------------------------------------------------- + +type chBlock struct { + hash string + number uint64 + parentHash string + blockTime time.Time + coinbase string + signer string + difficulty uint64 + totalDifficulty *big.Int + gasUsed uint64 + gasLimit uint64 + baseFee uint64 + txCount uint32 + uncleCount uint16 + uncleHash string + stateRoot string + txRoot string + receiptRoot string + logsBloom []byte + extraData []byte + mixDigest string + nonce uint64 + ingestedAt time.Time +} + +type chEvent struct { + hash string + peerID string + seenAt time.Time +} + +type chTx struct { + hash string + from string + to string + value *big.Int + gas uint64 + gasPrice *big.Int + gasFeeCap *big.Int + gasTipCap *big.Int + nonce uint64 + txType uint8 + firstSeen time.Time + ingestedAt time.Time +} + +type chPeer struct { + peerID string + name string + url string + caps []string + timeLastSeen time.Time +} + +// --- Database interface ---------------------------------------------------- + +func (c *ClickHouse) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time) { + if c.conn == nil { + return + } + if c.shouldWriteBlockEvents && peer != nil { + c.blockEvt.add(chEvent{hash: block.Hash().Hex(), peerID: peer.URLv4(), seenAt: tfs}) + } + if c.shouldWriteBlocks { + c.blocks.add(c.newBlock(block.Header(), td, tfs, len(block.Transactions()), len(block.Uncles()))) + } + if c.shouldWriteTransactions { + c.writeTxs(block.Transactions(), tfs) + } +} + +func (c *ClickHouse) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool) { + if c.conn == nil || !c.shouldWriteBlocks { + return + } + // A header alone carries no transaction/uncle counts; they default to 0 and + // are populated by the full-block (NewBlock) path when available. + for _, h := range headers { + c.blocks.add(c.newBlock(h, big.NewInt(0), tfs, 0, 0)) + } +} + +func (c *ClickHouse) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time) { + if c.conn == nil || !c.shouldWriteTransactions { + return + } + // The block row is written from the header path; here we only persist the + // transactions carried in the body (no read-modify-write on blocks). + txs, err := body.Transactions.Items() + if err != nil { + log.Error().Err(err).Str("hash", hash.Hex()).Msg("Failed to decode transactions from block body") + return + } + c.writeTxs(txs, tfs) +} + +func (c *ClickHouse) WriteBlockHashes(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time) { + if c.conn == nil || !c.shouldWriteBlockEvents || peer == nil { + return + } + for _, hash := range hashes { + c.blockEvt.add(chEvent{hash: hash.Hex(), peerID: peer.URLv4(), seenAt: tfs}) + } +} + +func (c *ClickHouse) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time) { + if c.conn == nil { + return + } + // Earliest first-seen is derived at query time from block_events (see the + // block_first_seen materialized view), so we only need to record the event. + // This mirrors the Datastore backend's first-block-event behavior. + if c.shouldWriteFirstBlockEvent && !c.shouldWriteBlockEvents && peer != nil { + c.blockEvt.add(chEvent{hash: hash.Hex(), peerID: peer.URLv4(), seenAt: tfsh}) + } +} + +func (c *ClickHouse) WriteTransactions(ctx context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time) { + if c.conn == nil { + return + } + if c.shouldWriteTransactions { + c.writeTxs(txs, tfs) + } + if c.shouldWriteTransactionEvents && peer != nil { + for _, tx := range txs { + c.txEvt.add(chEvent{hash: tx.Hash().Hex(), peerID: peer.URLv4(), seenAt: tfs}) + } + } +} + +func (c *ClickHouse) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.Time) { + if c.conn == nil || !c.shouldWritePeers { + return + } + for _, peer := range peers { + c.peers.add(chPeer{ + peerID: peer.ID().String(), + name: peer.Fullname(), + url: peer.Node().URLv4(), + caps: peer.Info().Caps, + timeLastSeen: tls, + }) + } +} + +// HasBlock reports whether the block already exists. It is called once per new +// block (not per event), so a lightweight indexed point lookup is cheap and +// preserves the parent-backfill behavior of the Datastore backend. +func (c *ClickHouse) HasBlock(ctx context.Context, hash common.Hash) bool { + if c.conn == nil { + return true + } + var exists uint8 + err := c.conn.QueryRow(ctx, "SELECT 1 FROM blocks WHERE hash = ? LIMIT 1", hash.Hex()).Scan(&exists) + return err == nil && exists == 1 +} + +func (c *ClickHouse) NodeList(ctx context.Context, limit int) ([]string, error) { + if c.conn == nil { + return []string{}, nil + } + rows, err := c.conn.Query(ctx, + "SELECT peer_id FROM block_events GROUP BY peer_id ORDER BY max(seen_at) DESC LIMIT ?", limit) + if err != nil { + return nil, err + } + defer func() { + if err := rows.Close(); err != nil { + log.Error().Err(err).Msg("Failed to close ClickHouse rows") + } + }() + + nodelist := []string{} + for rows.Next() { + var peerID string + if err := rows.Scan(&peerID); err != nil { + log.Error().Err(err).Msg("Failed to scan peer_id") + continue + } + nodelist = append(nodelist, peerID) + } + return nodelist, rows.Err() +} + +func (c *ClickHouse) MaxConcurrentWrites() int { return c.maxConcurrency } +func (c *ClickHouse) ShouldWriteBlocks() bool { return c.shouldWriteBlocks } +func (c *ClickHouse) ShouldWriteBlockEvents() bool { return c.shouldWriteBlockEvents } +func (c *ClickHouse) ShouldWriteFirstBlockEvent() bool { return c.shouldWriteFirstBlockEvent } +func (c *ClickHouse) ShouldWriteTransactions() bool { return c.shouldWriteTransactions } +func (c *ClickHouse) ShouldWriteTransactionEvents() bool { return c.shouldWriteTransactionEvents } +func (c *ClickHouse) ShouldWriteFirstTransactionEvent() bool { + return c.shouldWriteFirstTransactionEvent +} +func (c *ClickHouse) ShouldWritePeers() bool { return c.shouldWritePeers } + +// --- helpers --------------------------------------------------------------- + +func (c *ClickHouse) newBlock(h *types.Header, td *big.Int, tfs time.Time, txCount, uncleCount int) chBlock { + baseFee := uint64(0) + if h.BaseFee != nil { + baseFee = h.BaseFee.Uint64() + } + if td == nil { + td = big.NewInt(0) + } + // Recover the block signer from the header seal so signer-based analytics + // (double-signers, stolen/sealing blocks, reorgs) don't have to ecrecover on + // every query. Uses the same recovery as the sensor's block-signer validation + // and the data-analysis tools. Left empty when it can't be recovered. + var signer string + if sig, err := util.Ecrecover(h); err == nil { + signer = common.BytesToAddress(sig).Hex() + } + return chBlock{ + hash: h.Hash().Hex(), + number: h.Number.Uint64(), + parentHash: h.ParentHash.Hex(), + blockTime: time.Unix(int64(h.Time), 0).UTC(), + coinbase: h.Coinbase.Hex(), + signer: signer, + difficulty: h.Difficulty.Uint64(), + totalDifficulty: new(big.Int).Set(td), + gasUsed: h.GasUsed, + gasLimit: h.GasLimit, + baseFee: baseFee, + txCount: uint32(txCount), + uncleCount: uint16(uncleCount), + uncleHash: h.UncleHash.Hex(), + stateRoot: h.Root.Hex(), + txRoot: h.TxHash.Hex(), + receiptRoot: h.ReceiptHash.Hex(), + logsBloom: h.Bloom.Bytes(), + extraData: h.Extra, + mixDigest: h.MixDigest.Hex(), + nonce: h.Nonce.Uint64(), + ingestedAt: tfs, + } +} + +func (c *ClickHouse) writeTxs(txs []*types.Transaction, tfs time.Time) { + for _, tx := range txs { + var from, to string + chainID := tx.ChainId() + if chainID == nil || chainID.Sign() <= 0 { + chainID = c.chainID + } + if addr, err := types.Sender(types.LatestSignerForChainID(chainID), tx); err == nil { + from = addr.Hex() + } + if tx.To() != nil { + to = tx.To().Hex() + } + c.txs.add(chTx{ + hash: tx.Hash().Hex(), + from: from, + to: to, + value: new(big.Int).Set(tx.Value()), + gas: tx.Gas(), + gasPrice: new(big.Int).Set(tx.GasPrice()), + gasFeeCap: new(big.Int).Set(tx.GasFeeCap()), + gasTipCap: new(big.Int).Set(tx.GasTipCap()), + nonce: tx.Nonce(), + txType: tx.Type(), + firstSeen: tfs, + ingestedAt: tfs, + }) + } +} + +// --- batching -------------------------------------------------------------- + +// rowBatcher buffers rows and flushes them in bulk, either when the buffer +// reaches maxRows or on a fixed interval. add is non-blocking: when the buffer +// is full rows are dropped and counted (logged periodically) so the sensor hot +// path is never stalled by a slow database. The buffer is a fixed size. +type rowBatcher[T any] struct { + name string + in chan T + maxRows int + flush func(context.Context, []T) error + dropped atomic.Uint64 +} + +func newRowBatcher[T any](ctx context.Context, name string, maxRows int, flush func(context.Context, []T) error) *rowBatcher[T] { + b := &rowBatcher[T]{ + name: name, + in: make(chan T, maxRows*2), + maxRows: maxRows, + flush: flush, + } + go b.loop(ctx) + return b +} + +func (b *rowBatcher[T]) add(v T) { + select { + case b.in <- v: + default: + b.dropped.Add(1) + } +} + +func (b *rowBatcher[T]) loop(ctx context.Context) { + ticker := time.NewTicker(chFlushInterval) + defer ticker.Stop() + + buf := make([]T, 0, b.maxRows) + doFlush := func() { + if len(buf) == 0 { + return + } + if err := b.flush(ctx, buf); err != nil { + log.Error().Err(err).Str("table", b.name).Int("rows", len(buf)).Msg("ClickHouse batch insert failed") + } + buf = buf[:0] + } + + for { + select { + case <-ctx.Done(): + // Drain any buffered rows before exiting. + for { + select { + case v := <-b.in: + buf = append(buf, v) + if len(buf) >= b.maxRows { + doFlush() + } + default: + doFlush() + return + } + } + case v := <-b.in: + buf = append(buf, v) + if len(buf) >= b.maxRows { + doFlush() + } + case <-ticker.C: + doFlush() + if d := b.dropped.Swap(0); d > 0 { + log.Warn().Uint64("dropped", d).Str("table", b.name).Msg("ClickHouse batcher dropped rows (buffer full)") + } + } + } +} diff --git a/p2p/database/clickhouse_test.go b/p2p/database/clickhouse_test.go new file mode 100644 index 000000000..60d77ef25 --- /dev/null +++ b/p2p/database/clickhouse_test.go @@ -0,0 +1,134 @@ +package database + +import ( + "context" + "math/big" + "os" + "testing" + "time" + + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + "github.com/ethereum/go-ethereum/consensus/clique" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" +) + +// TestClickHouseWrites exercises the ClickHouse backend end-to-end against a +// real server. It is skipped unless POLYCLI_TEST_CLICKHOUSE_DSN is set, e.g. +// +// POLYCLI_TEST_CLICKHOUSE_DSN=clickhouse://localhost:19000/sensor go test ./p2p/database/ -run TestClickHouseWrites -v +// +// The target database must already have the schema from +// sensor-network-tools/clickhouse_schema.sql applied. +func TestClickHouseWrites(t *testing.T) { + dsn := os.Getenv("POLYCLI_TEST_CLICKHOUSE_DSN") + if dsn == "" { + t.Skip("POLYCLI_TEST_CLICKHOUSE_DSN not set; skipping ClickHouse integration test") + } + + ctx, cancel := context.WithCancel(context.Background()) + db := NewClickHouse(ctx, ClickHouseOptions{ + DSN: dsn, + SensorID: "test-sensor", + ChainID: 137, + MaxConcurrency: 10, + ShouldWriteBlocks: true, + ShouldWriteBlockEvents: true, + ShouldWriteTransactions: true, + ShouldWriteTransactionEvents: true, + ShouldWritePeers: true, + TTL: 14 * 24 * time.Hour, + }) + + now := time.Now().UTC() + priv, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("generate key: %v", err) + } + wantSigner := crypto.PubkeyToAddress(priv.PublicKey) + header := &types.Header{ + Number: big.NewInt(42), + Time: uint64(now.Unix()), + Difficulty: big.NewInt(7), + GasLimit: 30_000_000, + GasUsed: 21_000, + BaseFee: big.NewInt(1_000_000_000), + Extra: make([]byte, crypto.SignatureLength), + } + sig, err := crypto.Sign(clique.SealHash(header).Bytes(), priv) + if err != nil { + t.Fatalf("sign header: %v", err) + } + copy(header.Extra[len(header.Extra)-crypto.SignatureLength:], sig) + block := types.NewBlockWithHeader(header) + + tx := types.NewTx(&types.LegacyTx{ + Nonce: 1, + GasPrice: big.NewInt(2_000_000_000), + Gas: 21_000, + Value: big.NewInt(1), + }) + + db.WriteBlock(ctx, nil, block, big.NewInt(100), now) + db.WriteTransactions(ctx, nil, []*types.Transaction{tx}, now) + db.WritePeers(ctx, nil, now) // empty peer slice is fine; exercises the path + + // Trigger the drain flush and give it a moment to complete. + cancel() + time.Sleep(750 * time.Millisecond) + + conn, err := clickhouse.Open(mustParseDSN(t, dsn)) + if err != nil { + t.Fatalf("open verify conn: %v", err) + } + defer func() { + if err := conn.Close(); err != nil { + t.Errorf("close verify conn: %v", err) + } + }() + + checkCount(t, conn, "blocks", block.Hash().Hex()) + checkCount(t, conn, "transactions", tx.Hash().Hex()) + + // Verify the round-tripped block fields. + var ( + number uint64 + gasUsed uint64 + baseFee uint64 + signer string + ) + row := conn.QueryRow(context.Background(), + "SELECT number, gas_used, base_fee, signer FROM blocks WHERE hash = ? LIMIT 1", block.Hash().Hex()) + if err := row.Scan(&number, &gasUsed, &baseFee, &signer); err != nil { + t.Fatalf("scan block: %v", err) + } + if number != 42 || gasUsed != 21_000 || baseFee != 1_000_000_000 { + t.Fatalf("unexpected block fields: number=%d gas_used=%d base_fee=%d", number, gasUsed, baseFee) + } + if signer != wantSigner.Hex() { + t.Fatalf("signer mismatch: want %s got %s", wantSigner.Hex(), signer) + } +} + +func mustParseDSN(t *testing.T, dsn string) *clickhouse.Options { + t.Helper() + opts, err := clickhouse.ParseDSN(dsn) + if err != nil { + t.Fatalf("parse dsn: %v", err) + } + return opts +} + +func checkCount(t *testing.T, conn driver.Conn, table, hash string) { + t.Helper() + var count uint64 + // #nosec G202 -- table is a test-only constant, not user input + if err := conn.QueryRow(context.Background(), + "SELECT count() FROM "+table+" WHERE hash = ?", hash).Scan(&count); err != nil { + t.Fatalf("count %s: %v", table, err) + } + if count == 0 { + t.Fatalf("expected a row in %s for hash %s, got none", table, hash) + } +}