diff --git a/go.mod b/go.mod index ccd9885d0..5a2b0c263 100644 --- a/go.mod +++ b/go.mod @@ -40,7 +40,6 @@ require ( github.com/robfig/cron/v3 v3.0.1 github.com/shirou/gopsutil v3.21.11+incompatible github.com/spf13/viper v1.18.2 - github.com/stathat/consistent v1.0.0 go.etcd.io/etcd/client/v3 v3.5.13 go.uber.org/automaxprocs v1.5.3 golang.org/x/sync v0.8.0 @@ -88,19 +87,27 @@ require ( github.com/eapache/go-resiliency v1.6.0 // indirect github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect github.com/eapache/queue v1.1.0 // indirect + github.com/emicklei/go-restful/v3 v3.11.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/fsnotify/fsnotify v1.7.0 // indirect + github.com/fxamacker/cbor/v2 v2.7.0 // indirect github.com/gabriel-vasile/mimetype v1.4.3 // indirect github.com/gin-contrib/sse v0.1.0 // indirect github.com/go-logr/logr v1.4.2 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-ole/go-ole v1.2.6 // indirect + github.com/go-openapi/jsonpointer v0.19.6 // indirect + github.com/go-openapi/jsonreference v0.20.2 // indirect + github.com/go-openapi/swag v0.22.4 // indirect github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-zookeeper/zk v1.0.3 // indirect github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect github.com/golang/protobuf v1.5.4 // indirect github.com/golang/snappy v0.0.4 // indirect + github.com/google/gnostic-models v0.6.8 // indirect + github.com/google/go-cmp v0.6.0 // indirect github.com/google/go-querystring v1.1.0 // indirect + github.com/google/gofuzz v1.2.0 // indirect github.com/google/s2a-go v0.1.7 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect github.com/googleapis/gax-go/v2 v2.12.3 // indirect @@ -117,6 +124,7 @@ require ( github.com/jinzhu/copier v0.4.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect + github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kelindar/simd v1.1.2 // indirect github.com/klauspost/compress v1.17.7 // indirect @@ -126,6 +134,7 @@ require ( github.com/lithammer/shortuuid v3.0.0+incompatible // indirect github.com/magefile/mage v1.15.0 // indirect github.com/magiconair/properties v1.8.7 // indirect + github.com/mailru/easyjson v0.7.7 // indirect github.com/mattn/go-colorable v0.1.13 // indirect github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect github.com/minio/md5-simd v1.1.2 // indirect @@ -135,6 +144,7 @@ require ( github.com/modern-go/reflect2 v1.0.2 // indirect github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect github.com/mozillazg/go-httpheader v0.4.0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pelletier/go-toml/v2 v2.2.2 // indirect github.com/pierrec/lz4/v4 v4.1.21 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect @@ -156,6 +166,7 @@ require ( github.com/tklauser/go-sysconf v0.3.13 // indirect github.com/tklauser/numcpus v0.7.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect + github.com/x448/float16 v0.8.4 // indirect github.com/xdg-go/pbkdf2 v1.0.0 // indirect github.com/xdg-go/scram v1.1.2 // indirect github.com/xdg-go/stringprep v1.0.4 // indirect @@ -177,14 +188,25 @@ require ( golang.org/x/net v0.29.0 // indirect golang.org/x/oauth2 v0.23.0 // indirect golang.org/x/sys v0.25.0 // indirect + golang.org/x/term v0.24.0 // indirect golang.org/x/text v0.18.0 // indirect golang.org/x/time v0.5.0 // indirect google.golang.org/appengine/v2 v2.0.2 // indirect google.golang.org/genproto v0.0.0-20240213162025-012b6fc9bca9 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20240903143218-8af14fe29dc1 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240903143218-8af14fe29dc1 // indirect + gopkg.in/inf.v0 v0.9.1 // indirect + gopkg.in/yaml.v2 v2.4.0 // indirect gorm.io/gorm v1.25.8 // indirect - stathat.com/c/consistent v1.0.0 // indirect + k8s.io/api v0.31.2 // indirect + k8s.io/apimachinery v0.31.2 // indirect + k8s.io/client-go v0.31.2 // indirect + k8s.io/klog/v2 v2.130.1 // indirect + k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect + k8s.io/utils v0.0.0-20240711033017-18e509b52bc8 // indirect + sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect + sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect + sigs.k8s.io/yaml v1.4.0 // indirect ) require ( diff --git a/go.sum b/go.sum index 18d2a6495..af3515763 100644 --- a/go.sum +++ b/go.sum @@ -103,6 +103,8 @@ github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4A github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0= github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc= github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I= +github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g= +github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4= github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98= @@ -117,6 +119,8 @@ github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHk github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA= github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM= +github.com/fxamacker/cbor/v2 v2.7.0 h1:iM5WgngdRBanHcxugY4JySA0nk1wZorNOpTgCMedv5E= +github.com/fxamacker/cbor/v2 v2.7.0/go.mod h1:pxXPTn3joSm21Gbwsv0w9OSA2y1HFR9qXEeXQVeNoDQ= github.com/gabriel-vasile/mimetype v1.4.3 h1:in2uUcidCuFcDKtdcBxlR0rJ1+fsokWf+uqxgUFjbI0= github.com/gabriel-vasile/mimetype v1.4.3/go.mod h1:d8uq/6HKRL6CGdk+aubisF/M5GcPfT7nKyLpA0lbSSk= github.com/gin-contrib/gzip v1.0.1 h1:HQ8ENHODeLY7a4g1Au/46Z92bdGFl74OhxcZble9WJE= @@ -132,6 +136,13 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/go-openapi/jsonpointer v0.19.6 h1:eCs3fxoIi3Wh6vtgmLTOjdhSpiqphQ+DaPn38N2ZdrE= +github.com/go-openapi/jsonpointer v0.19.6/go.mod h1:osyAmYz/mB/C3I+WsTTSgw1ONzaLJoLCyoi6/zppojs= +github.com/go-openapi/jsonreference v0.20.2 h1:3sVjiK66+uXK/6oQ8xgcRKcFgQ5KXa2KvnJRumpMGbE= +github.com/go-openapi/jsonreference v0.20.2/go.mod h1:Bl1zwGIM8/wsvqjsOQLJ/SH+En5Ap4rVB5KVcIDZG2k= +github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14= +github.com/go-openapi/swag v0.22.4 h1:QLMzNJnMGPRNDCbySlcj1x01tzU8/9LTTL9hZZZogBU= +github.com/go-openapi/swag v0.22.4/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14= github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s= github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= @@ -150,6 +161,8 @@ github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGK github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA= github.com/go-redis/redismock/v9 v9.2.0 h1:ZrMYQeKPECZPjOj5u9eyOjg8Nnb0BS9lkVIZ6IpsKLw= github.com/go-redis/redismock/v9 v9.2.0/go.mod h1:18KHfGDK4Y6c2R0H38EUGWAdc7ZQS9gfYxc94k7rWT0= +github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= +github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/go-zookeeper/zk v1.0.3 h1:7M2kwOsc//9VeeFiPtf+uSJlVpU66x9Ba5+8XK7/TDg= github.com/go-zookeeper/zk v1.0.3/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw= github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU= @@ -179,6 +192,8 @@ github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/gnostic-models v0.6.8 h1:yo/ABAfM5IMRsS1VnXjTBvUb61tFIHozhlYvRgGre9I= +github.com/google/gnostic-models v0.6.8/go.mod h1:5n7qKqH0f5wFt+aWF8CW6pZLLNOfYuF5OpfBSENuI8U= github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= @@ -186,14 +201,19 @@ github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck= github.com/google/go-querystring v1.1.0 h1:AnCroh3fv4ZBgVIf1Iwtovgjaw/GiKJo8M8yD/fhyJ8= github.com/google/go-querystring v1.1.0/go.mod h1:Kcdr2DB4koayq7X8pmAG4sNG59So17icRSOU623lUBU= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0= +github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/martian/v3 v3.3.2 h1:IqNFLAmvJOgVlpdEBiQbDc2EwKW77amAycfTuWKdfvw= github.com/google/martian/v3 v3.3.2/go.mod h1:oBOf6HBosgwRXnUGWUB05QECsc6uvmMiJ3+6W4l/CUk= +github.com/google/pprof v0.0.0-20240525223248-4bfdf5a9a2af h1:kmjWCqn2qkEml422C2Rrd27c3VGxi6a/6HNq8QmHRKM= +github.com/google/pprof v0.0.0-20240525223248-4bfdf5a9a2af/go.mod h1:K1liHPHnj73Fdn/EKuT8nrFqBihUSKXoLYU0BuatOYo= github.com/google/s2a-go v0.1.7 h1:60BLSyTrOV4/haCDW4zb1guZItoSq8foHCXrAnjBo/o= github.com/google/s2a-go v0.1.7/go.mod h1:50CgR4k1jNlWBu4UfS4AcfhVe1r6pdZPygJ3R8F0Qdw= github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= @@ -244,6 +264,8 @@ github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/jonboulle/clockwork v0.4.0 h1:p4Cf1aMWXnXAUh8lVfewRBx1zaTSYKrKMF2g3ST4RZ4= github.com/jonboulle/clockwork v0.4.0/go.mod h1:xgRqUGwRcjKCO1vbZUEtSLrqKoPSsUpK7fnezOII0kc= +github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= +github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kelindar/bitmap v1.5.2 h1:XwX7CTvJtetQZ64zrOkApoZZHBJRkjE23NfqUALA/HE= @@ -285,6 +307,8 @@ github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg= github.com/magefile/mage v1.15.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY= github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= +github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= @@ -311,14 +335,18 @@ github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJ github.com/mozillazg/go-httpheader v0.2.1/go.mod h1:jJ8xECTlalr6ValeXYdOF8fFUISeBAdw6E61aqQma60= github.com/mozillazg/go-httpheader v0.4.0 h1:aBn6aRXtFzyDLZ4VIRLsZbbJloagQfMnCiYgOq6hK4w= github.com/mozillazg/go-httpheader v0.4.0/go.mod h1:PuT8h0pw6efvp8ZeUec1Rs7dwjK08bt6gKSReGMqtdA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE= github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU= github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU= +github.com/onsi/ginkgo/v2 v2.19.0 h1:9Cnnf7UHo57Hy3k6/m5k3dRfGTMXGvxhHFvkDTCTpvA= +github.com/onsi/ginkgo/v2 v2.19.0/go.mod h1:rlwLi9PilAFJ8jCg9UE1QP6VBpd6/xj3SRC0d6TU0To= github.com/onsi/gomega v1.25.0 h1:Vw7br2PCDYijJHSfBOWhov+8cAnUf8MfMaIOV323l6Y= github.com/onsi/gomega v1.25.0/go.mod h1:r+zV744Re+DiYCIPRlYOTxn0YkOLcAnW8k1xXdMPGhM= -github.com/openimsdk/gomake v0.0.15-alpha.2 h1:5Q8yl8ezy2yx+q8/ucU/t4kJnDfCzNOrkXcDACCqtyM= -github.com/openimsdk/gomake v0.0.15-alpha.2/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= +github.com/openimsdk/gomake v0.0.14-alpha.5 h1:VY9c5x515lTfmdhhPjMvR3BBRrRquAUCFsz7t7vbv7Y= +github.com/openimsdk/gomake v0.0.14-alpha.5/go.mod h1:PndCozNc2IsQIciyn9mvEblYWZwJmAI+06z94EY+csI= github.com/openimsdk/protocol v0.0.72-alpha.78 h1:n9HVj5olMPiGLF3Z4apPvvYzn2yOpyrsn2/YiAaIsxw= github.com/openimsdk/protocol v0.0.72-alpha.78/go.mod h1:WF7EuE55vQvpyUAzDXcqg+B+446xQyEba0X35lTINmw= github.com/openimsdk/tools v0.0.50-alpha.74 h1:yh10SiMiivMEjicEQg+QAsH4pvaO+4noMPdlw+ew0Kc= @@ -356,8 +384,8 @@ github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE= -github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= -github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= +github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= +github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc= github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= @@ -379,8 +407,6 @@ github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= github.com/spf13/viper v1.18.2 h1:LUXCnvUvSM6FXAsj6nnfc8Q2tp1dIgUfY9Kc8GsSOiQ= github.com/spf13/viper v1.18.2/go.mod h1:EKmWIqdnk5lOcmR72yw6hS+8OPYcwD0jteitLMVB+yk= -github.com/stathat/consistent v1.0.0 h1:ZFJ1QTRn8npNBKW065raSZ8xfOqhpb8vLOkfp4CcL/U= -github.com/stathat/consistent v1.0.0/go.mod h1:uajTPbgSygZBJ+V+0mY7meZ8i0XAcZs7AQ6V121XSxw= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= @@ -410,6 +436,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08= github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE= github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY= @@ -449,8 +477,8 @@ go.uber.org/atomic v1.9.0 h1:ECmE8Bn/WFTYwEW/bpKD3M8VtR/zQVbavAoalC1PYyE= go.uber.org/atomic v1.9.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8= go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0= -go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI= -go.uber.org/goleak v1.1.11/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60= @@ -527,6 +555,8 @@ golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9sn golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/term v0.24.0 h1:Mh5cbb+Zk2hqqXNO7S1iTjEphVL+jb8ZWaqh/g+JWkM= +golang.org/x/term v0.24.0/go.mod h1:lOBK/LVxemqiMij05LGJ0tzNr8xlmwBRJ81PX6wVLH8= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= @@ -548,6 +578,8 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -592,11 +624,14 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= +gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= +gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA= gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= @@ -607,7 +642,23 @@ gorm.io/gorm v1.25.8 h1:WAGEZ/aEcznN4D03laj8DKnehe1e9gYQAjW8xyPRdeo= gorm.io/gorm v1.25.8/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= +k8s.io/api v0.31.2 h1:3wLBbL5Uom/8Zy98GRPXpJ254nEFpl+hwndmk9RwmL0= +k8s.io/api v0.31.2/go.mod h1:bWmGvrGPssSK1ljmLzd3pwCQ9MgoTsRCuK35u6SygUk= +k8s.io/apimachinery v0.31.2 h1:i4vUt2hPK56W6mlT7Ry+AO8eEsyxMD1U44NR22CLTYw= +k8s.io/apimachinery v0.31.2/go.mod h1:rsPdaZJfTfLsNJSQzNHQvYoTmxhoOEofxtOsF3rtsMo= +k8s.io/client-go v0.31.2 h1:Y2F4dxU5d3AQj+ybwSMqQnpZH9F30//1ObxOKlTI9yc= +k8s.io/client-go v0.31.2/go.mod h1:NPa74jSVR/+eez2dFsEIHNa+3o09vtNaWwWwb1qSxSs= +k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk= +k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE= +k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 h1:BZqlfIlq5YbRMFko6/PM7FjZpUb45WallggurYhKGag= +k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340/go.mod h1:yD4MZYeKMBwQKVht279WycxKyM84kkAx2DPrTXaeb98= +k8s.io/utils v0.0.0-20240711033017-18e509b52bc8 h1:pUdcCO1Lk/tbT5ztQWOBi5HBgbBP1J8+AsQnQCKsi8A= +k8s.io/utils v0.0.0-20240711033017-18e509b52bc8/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= nullprogram.com/x/optparse v1.0.0/go.mod h1:KdyPE+Igbe0jQUrVfMqDMeJQIJZEuyV7pjYmp6pbG50= rsc.io/pdf v0.1.1/go.mod h1:n8OzWcQ6Sp37PL01nO98y4iUCRdTGarVfzxY20ICaU4= -stathat.com/c/consistent v1.0.0 h1:ezyc51EGcRPJUxfHGSgJjWzJdj3NiMU9pNfLNGiXV0c= -stathat.com/c/consistent v1.0.0/go.mod h1:QkzMWzcbB+yQBL2AttO6sgsQS/JSTapcDISJalmCDS0= +sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd h1:EDPBXCAspyGV4jQlpZSudPeMmr1bNJefnuqLsRAsHZo= +sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0= +sigs.k8s.io/structured-merge-diff/v4 v4.4.1 h1:150L+0vs/8DA78h1u02ooW1/fFq/Lwr+sGiqlzvrtq4= +sigs.k8s.io/structured-merge-diff/v4 v4.4.1/go.mod h1:N8hJocpFajUSSeSJ9bOZ77VzejKZaXsTtZo4/u7Io08= +sigs.k8s.io/yaml v1.4.0 h1:Mk1wCc2gy/F0THH0TAp1QYyJNzRm2KCLy3o5ASXVI5E= +sigs.k8s.io/yaml v1.4.0/go.mod h1:Ejl7/uTz7PSA4eKMyQCUTnhZYNmLIl+5c2lQPGR2BPY= diff --git a/internal/api/msg.go b/internal/api/msg.go index fc235354c..4fe950ffa 100644 --- a/internal/api/msg.go +++ b/internal/api/msg.go @@ -281,7 +281,7 @@ func (m *MessageApi) SendBusinessNotification(c *gin.Context) { IsSendMsg: false, ReliabilityLevel: 1, UnreadCount: false, - }), + }, nil), }, } respPb, err := m.Client.SendMsg(c, &sendMsgReq) diff --git a/internal/msggateway/ws_server.go b/internal/msggateway/ws_server.go index 208289e9e..a18ad0058 100644 --- a/internal/msggateway/ws_server.go +++ b/internal/msggateway/ws_server.go @@ -3,12 +3,13 @@ package msggateway import ( "context" "fmt" - "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "net/http" "sync" "sync/atomic" "time" + "github.com/openimsdk/open-im-server/v3/pkg/rpcli" + "github.com/go-playground/validator/v10" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" @@ -212,15 +213,19 @@ func (ws *WsServer) sendUserOnlineInfoToOtherNode(ctx context.Context, client *C if err != nil { return err } + if len(conns) == 0 || (len(conns) == 1 && ws.disCov.IsSelfNode(conns[0])) { + return nil + } + wg := errgroup.Group{} wg.SetLimit(concurrentRequest) // Online push user online message to other node for _, v := range conns { v := v - log.ZDebug(ctx, " sendUserOnlineInfoToOtherNode conn ", "target", v.Target()) - if v.Target() == ws.disCov.GetSelfConnTarget() { - log.ZDebug(ctx, "Filter out this node", "node", v.Target()) + log.ZDebug(ctx, "sendUserOnlineInfoToOtherNode conn") + if ws.disCov.IsSelfNode(v) { + log.ZDebug(ctx, "Filter out this node") continue } @@ -231,7 +236,7 @@ func (ws *WsServer) sendUserOnlineInfoToOtherNode(ctx context.Context, client *C PlatformID: int32(client.PlatformID), Token: client.token, }) if err != nil { - log.ZWarn(ctx, "MultiTerminalLoginCheck err", err, "node", v.Target()) + log.ZWarn(ctx, "MultiTerminalLoginCheck err", err) } return nil }) diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go index fce0297ee..b85848d9e 100644 --- a/internal/msgtransfer/online_history_msg_handler.go +++ b/internal/msgtransfer/online_history_msg_handler.go @@ -30,6 +30,7 @@ import ( "github.com/IBM/sarama" "github.com/go-redis/redis" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/openimsdk/open-im-server/v3/pkg/tools/batcher" "github.com/openimsdk/protocol/constant" @@ -38,7 +39,6 @@ import ( "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/mcontext" - "github.com/openimsdk/tools/mq/kafka" "github.com/openimsdk/tools/utils/stringutil" "google.golang.org/protobuf/proto" ) diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index d8836d54e..7892d0276 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -21,9 +21,9 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" pbmsg "github.com/openimsdk/protocol/msg" "github.com/openimsdk/tools/log" - "github.com/openimsdk/tools/mq/kafka" "google.golang.org/protobuf/proto" ) diff --git a/internal/push/offlinepush_handler.go b/internal/push/offlinepush_handler.go index 5c69da005..c685a188a 100644 --- a/internal/push/offlinepush_handler.go +++ b/internal/push/offlinepush_handler.go @@ -7,12 +7,12 @@ import ( "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush" "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/protocol/constant" pbpush "github.com/openimsdk/protocol/push" "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" - "github.com/openimsdk/tools/mq/kafka" "github.com/openimsdk/tools/utils/jsonutil" "google.golang.org/protobuf/proto" ) diff --git a/internal/push/onlinepusher.go b/internal/push/onlinepusher.go index 9521a84a0..2393c3567 100644 --- a/internal/push/onlinepusher.go +++ b/internal/push/onlinepusher.go @@ -2,6 +2,8 @@ package push import ( "context" + "sync" + "github.com/openimsdk/protocol/msggateway" "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/discovery" @@ -9,7 +11,6 @@ import ( "github.com/openimsdk/tools/utils/datautil" "golang.org/x/sync/errgroup" "google.golang.org/grpc" - "sync" ) type OnlinePusher interface { @@ -160,7 +161,7 @@ func (k *K8sStaticConsistentHash) GetConnsAndOnlinePush(ctx context.Context, msg } } log.ZDebug(ctx, "genUsers send hosts struct:", "usersHost", usersHost) - var usersConns = make(map[*grpc.ClientConn][]string) + var usersConns = make(map[grpc.ClientConnInterface][]string) for host, userIds := range usersHost { tconn, _ := k.disCov.GetConn(ctx, host) usersConns[tconn] = userIds diff --git a/internal/push/push_handler.go b/internal/push/push_handler.go index c82c8f898..e88e4b5a9 100644 --- a/internal/push/push_handler.go +++ b/internal/push/push_handler.go @@ -14,6 +14,7 @@ import ( "github.com/openimsdk/open-im-server/v3/internal/push/offlinepush/options" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/open-im-server/v3/pkg/common/webhook" "github.com/openimsdk/open-im-server/v3/pkg/msgprocessor" "github.com/openimsdk/open-im-server/v3/pkg/rpccache" @@ -25,7 +26,6 @@ import ( "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/log" "github.com/openimsdk/tools/mcontext" - "github.com/openimsdk/tools/mq/kafka" "github.com/openimsdk/tools/utils/datautil" "github.com/openimsdk/tools/utils/jsonutil" "github.com/openimsdk/tools/utils/timeutil" diff --git a/internal/rpc/auth/auth.go b/internal/rpc/auth/auth.go index 2bf08783d..a7055f6a5 100644 --- a/internal/rpc/auth/auth.go +++ b/internal/rpc/auth/auth.go @@ -192,7 +192,7 @@ func (s *authServer) forceKickOff(ctx context.Context, userID string, platformID return err } for _, v := range conns { - log.ZDebug(ctx, "forceKickOff", "conn", v.Target()) + log.ZDebug(ctx, "forceKickOff", "userID", userID, "platformID", platformID) client := msggateway.NewMsgGatewayClient(v) kickReq := &msggateway.KickUserOfflineReq{KickUserIDList: []string{userID}, PlatformID: platformID} _, err := client.KickUserOffline(ctx, kickReq) diff --git a/internal/rpc/conversation/notification.go b/internal/rpc/conversation/notification.go index c6368b916..370865c1a 100644 --- a/internal/rpc/conversation/notification.go +++ b/internal/rpc/conversation/notification.go @@ -16,21 +16,22 @@ package conversation import ( "context" + + "github.com/openimsdk/open-im-server/v3/pkg/notification" "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "github.com/openimsdk/protocol/msg" "github.com/openimsdk/open-im-server/v3/pkg/common/config" - "github.com/openimsdk/open-im-server/v3/pkg/notification" "github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/sdkws" ) type ConversationNotificationSender struct { - *rpcclient.NotificationSender + *notification.NotificationSender } func NewConversationNotificationSender(conf *config.Notification, msgClient *rpcli.MsgClient) *ConversationNotificationSender { - return &ConversationNotificationSender{rpcclient.NewNotificationSender(conf, rpcclient.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { + return &ConversationNotificationSender{notification.NewNotificationSender(conf, notification.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { return msgClient.SendMsg(ctx, req) }))} } diff --git a/internal/rpc/group/group.go b/internal/rpc/group/group.go index 0b03be6f5..2cd6c4f27 100644 --- a/internal/rpc/group/group.go +++ b/internal/rpc/group/group.go @@ -17,13 +17,14 @@ package group import ( "context" "fmt" - "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "math/big" "math/rand" "strconv" "strings" "time" + "github.com/openimsdk/open-im-server/v3/pkg/rpcli" + "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/common" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo" @@ -1129,7 +1130,7 @@ func (g *groupServer) SetGroupInfoEx(ctx context.Context, req *pbgroup.SetGroupI if notificationFlag { if req.Notification.Value != "" { - conversation := &pbconv.ConversationReq{ + conversation := &pbconversation.ConversationReq{ ConversationID: msgprocessor.GetConversationIDBySessionType(constant.ReadGroupChatType, req.GroupID), ConversationType: constant.ReadGroupChatType, GroupID: req.GroupID, diff --git a/internal/rpc/group/notification.go b/internal/rpc/group/notification.go index fb44bc312..ce42d0006 100644 --- a/internal/rpc/group/notification.go +++ b/internal/rpc/group/notification.go @@ -52,11 +52,11 @@ const ( func NewNotificationSender(db controller.GroupDatabase, config *Config, userClient *rpcli.UserClient, msgClient *rpcli.MsgClient, conversationClient *rpcli.ConversationClient) *NotificationSender { return &NotificationSender{ - NotificationSender: rpcclient.NewNotificationSender(&config.NotificationConfig, - rpcclient.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { + NotificationSender: notification.NewNotificationSender(&config.NotificationConfig, + notification.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { return msgClient.SendMsg(ctx, req) }), - rpcclient.WithUserRpcClient(userClient.GetUserInfo), + notification.WithUserRpcClient(userClient.GetUserInfo), ), getUsersInfo: func(ctx context.Context, userIDs []string) ([]common_user.CommonUser, error) { users, err := userClient.GetUsersInfo(ctx, userIDs) @@ -73,7 +73,7 @@ func NewNotificationSender(db controller.GroupDatabase, config *Config, userClie } type NotificationSender struct { - *rpcclient.NotificationSender + *notification.NotificationSender getUsersInfo func(ctx context.Context, userIDs []string) ([]common_user.CommonUser, error) db controller.GroupDatabase config *Config @@ -332,7 +332,7 @@ func (g *NotificationSender) GroupInfoSetNotification(ctx context.Context, tips return } g.setVersion(ctx, &tips.GroupMemberVersion, &tips.GroupMemberVersionID, database.GroupMemberVersionName, tips.Group.GroupID) - g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetNotification, tips, rpcclient.WithRpcGetUserName()) + g.Notification(ctx, mcontext.GetOpUserID(ctx), tips.Group.GroupID, constant.GroupInfoSetNotification, tips, notification.WithRpcGetUserName()) } func (g *NotificationSender) GroupInfoSetNameNotification(ctx context.Context, tips *sdkws.GroupInfoSetNameTips) { diff --git a/internal/rpc/msg/notification.go b/internal/rpc/msg/notification.go index d5604286a..0daafbe6c 100644 --- a/internal/rpc/msg/notification.go +++ b/internal/rpc/msg/notification.go @@ -23,11 +23,11 @@ import ( ) type MsgNotificationSender struct { - *rpcclient.NotificationSender + *notification.NotificationSender } -func NewMsgNotificationSender(config *Config, opts ...rpcclient.NotificationSenderOptions) *MsgNotificationSender { - return &MsgNotificationSender{rpcclient.NewNotificationSender(&config.NotificationConfig, opts...)} +func NewMsgNotificationSender(config *Config, opts ...notification.NotificationSenderOptions) *MsgNotificationSender { + return &MsgNotificationSender{notification.NewNotificationSender(&config.NotificationConfig, opts...)} } func (m *MsgNotificationSender) UserDeleteMsgsNotification(ctx context.Context, userID, conversationID string, seqs []int64) { @@ -48,3 +48,7 @@ func (m *MsgNotificationSender) MarkAsReadNotification(ctx context.Context, conv } m.NotificationWithSessionType(ctx, sendID, recvID, constant.HasReadReceipt, sessionType, tips) } + +func (m *MsgNotificationSender) StreamMsgNotification(ctx context.Context, sendID string, recvID string, sessionType int32, tips *sdkws.StreamMsgTips) { + m.NotificationWithSessionType(ctx, sendID, recvID, constant.StreamMsgNotification, sessionType, tips) +} diff --git a/internal/rpc/msg/server.go b/internal/rpc/msg/server.go index 74a6355d7..df5a72075 100644 --- a/internal/rpc/msg/server.go +++ b/internal/rpc/msg/server.go @@ -16,6 +16,8 @@ package msg import ( "context" + + "github.com/openimsdk/open-im-server/v3/pkg/notification" "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "github.com/openimsdk/open-im-server/v3/pkg/common/config" @@ -27,7 +29,6 @@ import ( "github.com/openimsdk/tools/db/redisutil" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/controller" - "github.com/openimsdk/open-im-server/v3/pkg/notification" "github.com/openimsdk/open-im-server/v3/pkg/rpccache" "github.com/openimsdk/protocol/constant" "github.com/openimsdk/protocol/conversation" @@ -63,7 +64,7 @@ type msgServer struct { GroupLocalCache *rpccache.GroupLocalCache // Local cache for group data. ConversationLocalCache *rpccache.ConversationLocalCache // Local cache for conversation data. Handlers MessageInterceptorChain // Chain of handlers for processing messages. - notificationSender *rpcclient.NotificationSender // RPC client for sending notifications. + notificationSender *notification.NotificationSender // RPC client for sending notifications. msgNotificationSender *MsgNotificationSender // RPC client for sending msg notifications. config *Config // Global configuration settings. webhookClient *webhook.Client @@ -132,8 +133,8 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg conversationClient: conversationClient, } - s.notificationSender = rpcclient.NewNotificationSender(&config.NotificationConfig, rpcclient.WithLocalSendMsg(s.SendMsg)) - s.msgNotificationSender = NewMsgNotificationSender(config, rpcclient.WithLocalSendMsg(s.SendMsg)) + s.notificationSender = notification.NewNotificationSender(&config.NotificationConfig, notification.WithLocalSendMsg(s.SendMsg)) + s.msgNotificationSender = NewMsgNotificationSender(config, notification.WithLocalSendMsg(s.SendMsg)) msg.RegisterMsgServer(server, s) diff --git a/internal/rpc/relation/notification.go b/internal/rpc/relation/notification.go index a34a4d322..caf2dafe1 100644 --- a/internal/rpc/relation/notification.go +++ b/internal/rpc/relation/notification.go @@ -16,6 +16,7 @@ package relation import ( "context" + "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "github.com/openimsdk/protocol/msg" @@ -36,7 +37,7 @@ import ( ) type FriendNotificationSender struct { - *rpcclient.NotificationSender + *notification.NotificationSender // Target not found err getUsersInfo func(ctx context.Context, userIDs []string) ([]common_user.CommonUser, error) // db controller @@ -89,7 +90,7 @@ func WithRpcFunc( func NewFriendNotificationSender(conf *config.Notification, msgClient *rpcli.MsgClient, opts ...friendNotificationSenderOptions) *FriendNotificationSender { f := &FriendNotificationSender{ - NotificationSender: rpcclient.NewNotificationSender(conf, rpcclient.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { + NotificationSender: notification.NewNotificationSender(conf, notification.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { return msgClient.SendMsg(ctx, req) })), } diff --git a/internal/rpc/third/s3.go b/internal/rpc/third/s3.go index 8796fe824..97206dd6d 100644 --- a/internal/rpc/third/s3.go +++ b/internal/rpc/third/s3.go @@ -19,11 +19,12 @@ import ( "encoding/base64" "encoding/hex" "encoding/json" - "github.com/openimsdk/open-im-server/v3/pkg/authverify" "path" "strconv" "time" + "github.com/openimsdk/open-im-server/v3/pkg/authverify" + "github.com/google/uuid" "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" @@ -37,7 +38,10 @@ import ( ) func (t *thirdServer) PartLimit(ctx context.Context, req *third.PartLimitReq) (*third.PartLimitResp, error) { - limit := t.s3dataBase.PartLimit() + limit, err := t.s3dataBase.PartLimit() + if err != nil { + return nil, err + } return &third.PartLimitResp{ MinPartSize: limit.MinPartSize, MaxPartSize: limit.MaxPartSize, diff --git a/internal/rpc/user/notification.go b/internal/rpc/user/notification.go index 03fdf95bd..4fb214f74 100644 --- a/internal/rpc/user/notification.go +++ b/internal/rpc/user/notification.go @@ -16,6 +16,7 @@ package user import ( "context" + "github.com/openimsdk/open-im-server/v3/pkg/rpcli" "github.com/openimsdk/protocol/msg" @@ -29,7 +30,7 @@ import ( ) type UserNotificationSender struct { - *rpcclient.NotificationSender + *notification.NotificationSender getUsersInfo func(ctx context.Context, userIDs []string) ([]common_user.CommonUser, error) // db controller db controller.UserDatabase @@ -63,7 +64,7 @@ func WithUserFunc( func NewUserNotificationSender(config *Config, msgClient *rpcli.MsgClient, opts ...userNotificationSenderOptions) *UserNotificationSender { f := &UserNotificationSender{ - NotificationSender: rpcclient.NewNotificationSender(&config.NotificationConfig, rpcclient.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { + NotificationSender: notification.NewNotificationSender(&config.NotificationConfig, notification.WithRpcClient(func(ctx context.Context, req *msg.SendMsgReq) (*msg.SendMsgResp, error) { return msgClient.SendMsg(ctx, req) })), } diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go index e2b9f6c8d..12234233b 100644 --- a/pkg/common/config/config.go +++ b/pkg/common/config/config.go @@ -18,9 +18,9 @@ import ( "strings" "time" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/tools/db/mongoutil" "github.com/openimsdk/tools/db/redisutil" - "github.com/openimsdk/tools/mq/kafka" "github.com/openimsdk/tools/s3/aws" "github.com/openimsdk/tools/s3/cos" "github.com/openimsdk/tools/s3/kodo" diff --git a/pkg/common/discoveryregister/discoveryregister.go b/pkg/common/discoveryregister/discoveryregister.go index f7175491c..740d27eac 100644 --- a/pkg/common/discoveryregister/discoveryregister.go +++ b/pkg/common/discoveryregister/discoveryregister.go @@ -18,17 +18,22 @@ import ( "time" "github.com/openimsdk/open-im-server/v3/pkg/common/config" - "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/kubernetes" "github.com/openimsdk/tools/discovery" "github.com/openimsdk/tools/discovery/etcd" + "github.com/openimsdk/tools/discovery/kubernetes" "github.com/openimsdk/tools/errs" + "google.golang.org/grpc" ) // NewDiscoveryRegister creates a new service discovery and registry client based on the provided environment type. func NewDiscoveryRegister(discovery *config.Discovery, share *config.Share, watchNames []string) (discovery.SvcDiscoveryRegistry, error) { switch discovery.Enable { case "k8s": - return kubernetes.NewK8sDiscoveryRegister(share.RpcRegisterName.MessageGateway) + return kubernetes.NewKubernetesConnManager("default", + grpc.WithDefaultCallOptions( + grpc.MaxCallSendMsgSize(1024*1024*20), + ), + ) case "etcd": return etcd.NewSvcDiscoveryRegistry( discovery.Etcd.RootDirectory, diff --git a/pkg/common/discoveryregister/kubernetes/doc.go b/pkg/common/discoveryregister/kubernetes/doc.go deleted file mode 100644 index 8615caa6b..000000000 --- a/pkg/common/discoveryregister/kubernetes/doc.go +++ /dev/null @@ -1,15 +0,0 @@ -// Copyright © 2024 OpenIM. All rights reserved. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package kubernetes // import "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/kubernetes" diff --git a/pkg/common/discoveryregister/kubernetes/kubernetes.go b/pkg/common/discoveryregister/kubernetes/kubernetes.go deleted file mode 100644 index f1ce0bbdc..000000000 --- a/pkg/common/discoveryregister/kubernetes/kubernetes.go +++ /dev/null @@ -1,199 +0,0 @@ -// Copyright © 2023 OpenIM. All rights reserved. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package kubernetes - -import ( - "context" - "errors" - "fmt" - "os" - "strconv" - "strings" - - "github.com/openimsdk/tools/discovery" - "github.com/openimsdk/tools/log" - "github.com/stathat/consistent" - "google.golang.org/grpc" -) - -// K8sDR represents the Kubernetes service discovery and registration client. -type K8sDR struct { - options []grpc.DialOption - rpcRegisterAddr string - gatewayHostConsistent *consistent.Consistent - gatewayName string -} - -func NewK8sDiscoveryRegister(gatewayName string) (discovery.SvcDiscoveryRegistry, error) { - gatewayConsistent := consistent.New() - gatewayHosts := getMsgGatewayHost(context.Background(), gatewayName) - for _, v := range gatewayHosts { - gatewayConsistent.Add(v) - } - return &K8sDR{gatewayHostConsistent: gatewayConsistent}, nil -} - -func (cli *K8sDR) Register(serviceName, host string, port int, opts ...grpc.DialOption) error { - if serviceName != cli.gatewayName { - cli.rpcRegisterAddr = serviceName - } else { - cli.rpcRegisterAddr = getSelfHost(context.Background(), cli.gatewayName) - } - - return nil -} - -func (cli *K8sDR) UnRegister() error { - - return nil -} - -func (cli *K8sDR) CreateRpcRootNodes(serviceNames []string) error { - - return nil -} - -func (cli *K8sDR) RegisterConf2Registry(key string, conf []byte) error { - - return nil -} - -func (cli *K8sDR) GetConfFromRegistry(key string) ([]byte, error) { - - return nil, nil -} - -func (cli *K8sDR) GetUserIdHashGatewayHost(ctx context.Context, userId string) (string, error) { - host, err := cli.gatewayHostConsistent.Get(userId) - if err != nil { - log.ZError(ctx, "GetUserIdHashGatewayHost error", err) - } - return host, err -} - -func getSelfHost(ctx context.Context, gatewayName string) string { - port := 88 - instance := "openimserver" - selfPodName := os.Getenv("MY_POD_NAME") - ns := os.Getenv("MY_POD_NAMESPACE") - statefuleIndex := 0 - gatewayEnds := strings.Split(gatewayName, ":") - if len(gatewayEnds) != 2 { - log.ZError(ctx, "msggateway RpcRegisterName is error:config.RpcRegisterName.OpenImMessageGatewayName", errors.New("config error")) - } else { - port, _ = strconv.Atoi(gatewayEnds[1]) - } - podInfo := strings.Split(selfPodName, "-") - instance = podInfo[0] - count := len(podInfo) - statefuleIndex, _ = strconv.Atoi(podInfo[count-1]) - host := fmt.Sprintf("%s-openim-msggateway-%d.%s-openim-msggateway-headless.%s.svc.cluster.local:%d", instance, statefuleIndex, instance, ns, port) - return host -} - -// like openimserver-openim-msggateway-0.openimserver-openim-msggateway-headless.openim-lin.svc.cluster.local:88. -// Replica set in kubernetes environment -func getMsgGatewayHost(ctx context.Context, gatewayName string) []string { - port := 88 - instance := "openimserver" - selfPodName := os.Getenv("MY_POD_NAME") - replicas := os.Getenv("MY_MSGGATEWAY_REPLICACOUNT") - ns := os.Getenv("MY_POD_NAMESPACE") - gatewayEnds := strings.Split(gatewayName, ":") - if len(gatewayEnds) != 2 { - log.ZError(ctx, "msggateway RpcRegisterName is error:config.RpcRegisterName.OpenImMessageGatewayName", errors.New("config error")) - } else { - port, _ = strconv.Atoi(gatewayEnds[1]) - } - nReplicas, _ := strconv.Atoi(replicas) - podInfo := strings.Split(selfPodName, "-") - instance = podInfo[0] - var ret []string - for i := 0; i < nReplicas; i++ { - host := fmt.Sprintf("%s-openim-msggateway-%d.%s-openim-msggateway-headless.%s.svc.cluster.local:%d", instance, i, instance, ns, port) - ret = append(ret, host) - } - log.ZDebug(ctx, "getMsgGatewayHost", "instance", instance, "selfPodName", selfPodName, "replicas", replicas, "ns", ns, "ret", ret) - return ret -} - -// GetConns returns the gRPC client connections to the specified service. -func (cli *K8sDR) GetConns(ctx context.Context, serviceName string, opts ...grpc.DialOption) ([]*grpc.ClientConn, error) { - - // This conditional checks if the serviceName is not the OpenImMessageGatewayName. - // It seems to handle a special case for the OpenImMessageGateway. - if serviceName != cli.gatewayName { - // DialContext creates a client connection to the given target (serviceName) using the specified context. - // 'cli.options' are likely default or common options for all connections in this struct. - // 'opts...' allows for additional gRPC dial options to be passed and used. - conn, err := grpc.DialContext(ctx, serviceName, append(cli.options, opts...)...) - - // The function returns a slice of client connections with the new connection, or an error if occurred. - return []*grpc.ClientConn{conn}, err - } else { - // This block is executed if the serviceName is OpenImMessageGatewayName. - // 'ret' will accumulate the connections to return. - var ret []*grpc.ClientConn - - // getMsgGatewayHost presumably retrieves hosts for the message gateway service. - // The context is passed, likely for cancellation and timeout control. - gatewayHosts := getMsgGatewayHost(ctx, cli.gatewayName) - - // Iterating over the retrieved gateway hosts. - for _, host := range gatewayHosts { - // Establishes a connection to each host. - // Again, appending cli.options with any additional opts provided. - conn, err := grpc.DialContext(ctx, host, append(cli.options, opts...)...) - - // If there's an error while dialing any host, the function returns immediately with the error. - if err != nil { - return nil, err - } else { - // If the connection is successful, it is added to the 'ret' slice. - ret = append(ret, conn) - } - } - // After all hosts are processed, the slice of connections is returned. - return ret, nil - } -} - -func (cli *K8sDR) GetConn(ctx context.Context, serviceName string, opts ...grpc.DialOption) (*grpc.ClientConn, error) { - - return grpc.DialContext(ctx, serviceName, append(cli.options, opts...)...) -} - -func (cli *K8sDR) GetSelfConnTarget() string { - - return cli.rpcRegisterAddr -} - -func (cli *K8sDR) AddOption(opts ...grpc.DialOption) { - cli.options = append(cli.options, opts...) -} - -func (cli *K8sDR) CloseConn(conn *grpc.ClientConn) { - conn.Close() -} - -// do not use this method for call rpc. -func (cli *K8sDR) GetClientLocalConns() map[string][]*grpc.ClientConn { - log.ZError(context.Background(), "should not call this function!", nil) - return nil -} - -func (cli *K8sDR) Close() { - -} diff --git a/pkg/common/startrpc/start.go b/pkg/common/startrpc/start.go index 3d4394c51..6b6619914 100644 --- a/pkg/common/startrpc/start.go +++ b/pkg/common/startrpc/start.go @@ -167,6 +167,7 @@ func Start[T any](ctx context.Context, discovery *conf.Discovery, prometheusConf } err = client.Register( + ctx, rpcRegisterName, registerIP, port, diff --git a/pkg/common/storage/controller/msg.go b/pkg/common/storage/controller/msg.go index b92f9b510..e2df028ec 100644 --- a/pkg/common/storage/controller/msg.go +++ b/pkg/common/storage/controller/msg.go @@ -22,6 +22,7 @@ import ( "strings" "time" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/tools/utils/jsonutil" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database" @@ -38,7 +39,6 @@ import ( "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" - "github.com/openimsdk/tools/mq/kafka" "github.com/openimsdk/tools/utils/datautil" ) diff --git a/pkg/common/storage/controller/msg_transfer.go b/pkg/common/storage/controller/msg_transfer.go index f4c0c6270..db62ec289 100644 --- a/pkg/common/storage/controller/msg_transfer.go +++ b/pkg/common/storage/controller/msg_transfer.go @@ -2,7 +2,9 @@ package controller import ( "context" + "github.com/openimsdk/open-im-server/v3/pkg/common/convert" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/protocol/constant" "github.com/openimsdk/tools/utils/datautil" @@ -14,7 +16,6 @@ import ( "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" - "github.com/openimsdk/tools/mq/kafka" "go.mongodb.org/mongo-driver/mongo" ) diff --git a/pkg/common/storage/controller/push.go b/pkg/common/storage/controller/push.go index 91ef126fe..a805eaf00 100644 --- a/pkg/common/storage/controller/push.go +++ b/pkg/common/storage/controller/push.go @@ -19,10 +19,10 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/protocol/push" "github.com/openimsdk/protocol/sdkws" "github.com/openimsdk/tools/log" - "github.com/openimsdk/tools/mq/kafka" ) type PushDatabase interface { diff --git a/pkg/common/storage/controller/s3.go b/pkg/common/storage/controller/s3.go index 6693d2dde..30d8d20ec 100644 --- a/pkg/common/storage/controller/s3.go +++ b/pkg/common/storage/controller/s3.go @@ -30,7 +30,7 @@ import ( ) type S3Database interface { - PartLimit() *s3.PartLimit + PartLimit() (*s3.PartLimit, error) PartSize(ctx context.Context, size int64) (int64, error) AuthSign(ctx context.Context, uploadID string, partNumbers []int) (*s3.AuthSignResult, error) InitiateMultipartUpload(ctx context.Context, hash string, size int64, expire time.Duration, maxParts int) (*cont.InitiateUploadResult, error) @@ -65,7 +65,7 @@ func (s *s3Database) PartSize(ctx context.Context, size int64) (int64, error) { return s.s3.PartSize(ctx, size) } -func (s *s3Database) PartLimit() *s3.PartLimit { +func (s *s3Database) PartLimit() (*s3.PartLimit, error) { return s.s3.PartLimit() } diff --git a/pkg/common/storage/kafka/config.go b/pkg/common/storage/kafka/config.go new file mode 100644 index 000000000..1c9c4b0a0 --- /dev/null +++ b/pkg/common/storage/kafka/config.go @@ -0,0 +1,33 @@ +// Copyright © 2024 OpenIM open source community. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +type TLSConfig struct { + EnableTLS bool `yaml:"enableTLS"` + CACrt string `yaml:"caCrt"` + ClientCrt string `yaml:"clientCrt"` + ClientKey string `yaml:"clientKey"` + ClientKeyPwd string `yaml:"clientKeyPwd"` + InsecureSkipVerify bool `yaml:"insecureSkipVerify"` +} + +type Config struct { + Username string `yaml:"username"` + Password string `yaml:"password"` + ProducerAck string `yaml:"producerAck"` + CompressType string `yaml:"compressType"` + Addr []string `yaml:"addr"` + TLS TLSConfig `yaml:"tls"` +} diff --git a/pkg/common/storage/kafka/consumer_group.go b/pkg/common/storage/kafka/consumer_group.go new file mode 100644 index 000000000..f0e84bbc9 --- /dev/null +++ b/pkg/common/storage/kafka/consumer_group.go @@ -0,0 +1,68 @@ +// Copyright © 2023 OpenIM. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "context" + "errors" + + "github.com/IBM/sarama" + "github.com/openimsdk/tools/log" +) + +type MConsumerGroup struct { + sarama.ConsumerGroup + groupID string + topics []string +} + +func NewMConsumerGroup(conf *Config, groupID string, topics []string, autoCommitEnable bool) (*MConsumerGroup, error) { + config, err := BuildConsumerGroupConfig(conf, sarama.OffsetNewest, autoCommitEnable) + if err != nil { + return nil, err + } + group, err := NewConsumerGroup(config, conf.Addr, groupID) + if err != nil { + return nil, err + } + return &MConsumerGroup{ + ConsumerGroup: group, + groupID: groupID, + topics: topics, + }, nil +} + +func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) context.Context { + return GetContextWithMQHeader(cMsg.Headers) +} + +func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler) { + for { + err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler) + if errors.Is(err, sarama.ErrClosedConsumerGroup) { + return + } + if errors.Is(err, context.Canceled) { + return + } + if err != nil { + log.ZWarn(ctx, "consume err", err, "topic", mc.topics, "groupID", mc.groupID) + } + } +} + +func (mc *MConsumerGroup) Close() error { + return mc.ConsumerGroup.Close() +} diff --git a/pkg/common/storage/kafka/producer.go b/pkg/common/storage/kafka/producer.go new file mode 100644 index 000000000..5f6be29ed --- /dev/null +++ b/pkg/common/storage/kafka/producer.go @@ -0,0 +1,82 @@ +// Copyright © 2023 OpenIM. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "context" + "github.com/IBM/sarama" + "github.com/openimsdk/tools/errs" + "google.golang.org/protobuf/proto" +) + +// Producer represents a Kafka producer. +type Producer struct { + addr []string + topic string + config *sarama.Config + producer sarama.SyncProducer +} + +func NewKafkaProducer(config *sarama.Config, addr []string, topic string) (*Producer, error) { + producer, err := NewProducer(config, addr) + if err != nil { + return nil, err + } + return &Producer{ + addr: addr, + topic: topic, + config: config, + producer: producer, + }, nil +} + +// SendMessage sends a message to the Kafka topic configured in the Producer. +func (p *Producer) SendMessage(ctx context.Context, key string, msg proto.Message) (int32, int64, error) { + // Marshal the protobuf message + bMsg, err := proto.Marshal(msg) + if err != nil { + return 0, 0, errs.WrapMsg(err, "kafka proto Marshal err") + } + if len(bMsg) == 0 { + return 0, 0, errs.WrapMsg(errEmptyMsg, "kafka proto Marshal err") + } + + // Prepare Kafka message + kMsg := &sarama.ProducerMessage{ + Topic: p.topic, + Key: sarama.StringEncoder(key), + Value: sarama.ByteEncoder(bMsg), + } + + // Validate message key and value + if kMsg.Key.Length() == 0 || kMsg.Value.Length() == 0 { + return 0, 0, errs.Wrap(errEmptyMsg) + } + + // Attach context metadata as headers + header, err := GetMQHeaderWithContext(ctx) + if err != nil { + return 0, 0, err + } + kMsg.Headers = header + + // Send the message + partition, offset, err := p.producer.SendMessage(kMsg) + if err != nil { + return 0, 0, errs.WrapMsg(err, "p.producer.SendMessage error") + } + + return partition, offset, nil +} diff --git a/pkg/common/storage/kafka/sarama.go b/pkg/common/storage/kafka/sarama.go new file mode 100644 index 000000000..23220b4d0 --- /dev/null +++ b/pkg/common/storage/kafka/sarama.go @@ -0,0 +1,85 @@ +package kafka + +import ( + "bytes" + "strings" + + "github.com/IBM/sarama" + "github.com/openimsdk/tools/errs" +) + +func BuildConsumerGroupConfig(conf *Config, initial int64, autoCommitEnable bool) (*sarama.Config, error) { + kfk := sarama.NewConfig() + kfk.Version = sarama.V2_0_0_0 + kfk.Consumer.Offsets.Initial = initial + kfk.Consumer.Offsets.AutoCommit.Enable = autoCommitEnable + kfk.Consumer.Return.Errors = false + if conf.Username != "" || conf.Password != "" { + kfk.Net.SASL.Enable = true + kfk.Net.SASL.User = conf.Username + kfk.Net.SASL.Password = conf.Password + } + if conf.TLS.EnableTLS { + tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify) + if err != nil { + return nil, err + } + kfk.Net.TLS.Config = tls + kfk.Net.TLS.Enable = true + } + return kfk, nil +} + +func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error) { + cg, err := sarama.NewConsumerGroup(addr, groupID, conf) + if err != nil { + return nil, errs.WrapMsg(err, "NewConsumerGroup failed", "addr", addr, "groupID", groupID, "conf", *conf) + } + return cg, nil +} + +func BuildProducerConfig(conf Config) (*sarama.Config, error) { + kfk := sarama.NewConfig() + kfk.Producer.Return.Successes = true + kfk.Producer.Return.Errors = true + kfk.Producer.Partitioner = sarama.NewHashPartitioner + if conf.Username != "" || conf.Password != "" { + kfk.Net.SASL.Enable = true + kfk.Net.SASL.User = conf.Username + kfk.Net.SASL.Password = conf.Password + } + switch strings.ToLower(conf.ProducerAck) { + case "no_response": + kfk.Producer.RequiredAcks = sarama.NoResponse + case "wait_for_local": + kfk.Producer.RequiredAcks = sarama.WaitForLocal + case "wait_for_all": + kfk.Producer.RequiredAcks = sarama.WaitForAll + default: + kfk.Producer.RequiredAcks = sarama.WaitForAll + } + if conf.CompressType == "" { + kfk.Producer.Compression = sarama.CompressionNone + } else { + if err := kfk.Producer.Compression.UnmarshalText(bytes.ToLower([]byte(conf.CompressType))); err != nil { + return nil, errs.WrapMsg(err, "UnmarshalText failed", "compressType", conf.CompressType) + } + } + if conf.TLS.EnableTLS { + tls, err := newTLSConfig(conf.TLS.ClientCrt, conf.TLS.ClientKey, conf.TLS.CACrt, []byte(conf.TLS.ClientKeyPwd), conf.TLS.InsecureSkipVerify) + if err != nil { + return nil, err + } + kfk.Net.TLS.Config = tls + kfk.Net.TLS.Enable = true + } + return kfk, nil +} + +func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error) { + producer, err := sarama.NewSyncProducer(addr, conf) + if err != nil { + return nil, errs.WrapMsg(err, "NewSyncProducer failed", "addr", addr, "conf", *conf) + } + return producer, nil +} diff --git a/pkg/common/storage/kafka/tls.go b/pkg/common/storage/kafka/tls.go new file mode 100644 index 000000000..00c89dcc1 --- /dev/null +++ b/pkg/common/storage/kafka/tls.go @@ -0,0 +1,83 @@ +// Copyright © 2024 OpenIM open source community. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "crypto/tls" + "crypto/x509" + "encoding/pem" + "os" + + "github.com/openimsdk/tools/errs" +) + +// decryptPEM decrypts a PEM block using a password. +func decryptPEM(data []byte, passphrase []byte) ([]byte, error) { + if len(passphrase) == 0 { + return data, nil + } + b, _ := pem.Decode(data) + d, err := x509.DecryptPEMBlock(b, passphrase) + if err != nil { + return nil, errs.WrapMsg(err, "DecryptPEMBlock failed") + } + return pem.EncodeToMemory(&pem.Block{ + Type: b.Type, + Bytes: d, + }), nil +} + +func readEncryptablePEMBlock(path string, pwd []byte) ([]byte, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, errs.WrapMsg(err, "ReadFile failed", "path", path) + } + return decryptPEM(data, pwd) +} + +// newTLSConfig setup the TLS config from general config file. +func newTLSConfig(clientCertFile, clientKeyFile, caCertFile string, keyPwd []byte, insecureSkipVerify bool) (*tls.Config, error) { + var tlsConfig tls.Config + if clientCertFile != "" && clientKeyFile != "" { + certPEMBlock, err := os.ReadFile(clientCertFile) + if err != nil { + return nil, errs.WrapMsg(err, "ReadFile failed", "clientCertFile", clientCertFile) + } + keyPEMBlock, err := readEncryptablePEMBlock(clientKeyFile, keyPwd) + if err != nil { + return nil, err + } + + cert, err := tls.X509KeyPair(certPEMBlock, keyPEMBlock) + if err != nil { + return nil, errs.WrapMsg(err, "X509KeyPair failed") + } + tlsConfig.Certificates = []tls.Certificate{cert} + } + + if caCertFile != "" { + caCert, err := os.ReadFile(caCertFile) + if err != nil { + return nil, errs.WrapMsg(err, "ReadFile failed", "caCertFile", caCertFile) + } + caCertPool := x509.NewCertPool() + if ok := caCertPool.AppendCertsFromPEM(caCert); !ok { + return nil, errs.New("AppendCertsFromPEM failed") + } + tlsConfig.RootCAs = caCertPool + } + tlsConfig.InsecureSkipVerify = insecureSkipVerify + return &tlsConfig, nil +} diff --git a/pkg/common/storage/kafka/util.go b/pkg/common/storage/kafka/util.go new file mode 100644 index 000000000..61abe5450 --- /dev/null +++ b/pkg/common/storage/kafka/util.go @@ -0,0 +1,34 @@ +package kafka + +import ( + "context" + "errors" + "github.com/IBM/sarama" + "github.com/openimsdk/protocol/constant" + "github.com/openimsdk/tools/mcontext" +) + +var errEmptyMsg = errors.New("kafka binary msg is empty") + +// GetMQHeaderWithContext extracts message queue headers from the context. +func GetMQHeaderWithContext(ctx context.Context) ([]sarama.RecordHeader, error) { + operationID, opUserID, platform, connID, err := mcontext.GetCtxInfos(ctx) + if err != nil { + return nil, err + } + return []sarama.RecordHeader{ + {Key: []byte(constant.OperationID), Value: []byte(operationID)}, + {Key: []byte(constant.OpUserID), Value: []byte(opUserID)}, + {Key: []byte(constant.OpUserPlatform), Value: []byte(platform)}, + {Key: []byte(constant.ConnID), Value: []byte(connID)}, + }, nil +} + +// GetContextWithMQHeader creates a context from message queue headers. +func GetContextWithMQHeader(header []*sarama.RecordHeader) context.Context { + var values []string + for _, recordHeader := range header { + values = append(values, string(recordHeader.Value)) + } + return mcontext.WithMustInfoCtx(values) // Attach extracted values to context +} diff --git a/pkg/common/storage/kafka/verify.go b/pkg/common/storage/kafka/verify.go new file mode 100644 index 000000000..0a09eed4e --- /dev/null +++ b/pkg/common/storage/kafka/verify.go @@ -0,0 +1,79 @@ +// Copyright © 2024 OpenIM open source community. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "context" + "fmt" + + "github.com/IBM/sarama" + "github.com/openimsdk/tools/errs" +) + +func CheckTopics(ctx context.Context, conf *Config, topics []string) error { + kfk, err := BuildConsumerGroupConfig(conf, sarama.OffsetNewest, false) + if err != nil { + return err + } + cli, err := sarama.NewClient(conf.Addr, kfk) + if err != nil { + return errs.WrapMsg(err, "NewClient failed", "config: ", fmt.Sprintf("%+v", conf)) + } + defer cli.Close() + + existingTopics, err := cli.Topics() + if err != nil { + return errs.WrapMsg(err, "Failed to list topics") + } + + existingTopicsMap := make(map[string]bool) + for _, t := range existingTopics { + existingTopicsMap[t] = true + } + + for _, topic := range topics { + if !existingTopicsMap[topic] { + return errs.New("topic not exist", "topic", topic).Wrap() + } + } + return nil +} + +func CheckHealth(ctx context.Context, conf *Config) error { + kfk, err := BuildConsumerGroupConfig(conf, sarama.OffsetNewest, false) + if err != nil { + return err + } + cli, err := sarama.NewClient(conf.Addr, kfk) + if err != nil { + return errs.WrapMsg(err, "NewClient failed", "config: ", fmt.Sprintf("%+v", conf)) + } + defer cli.Close() + + // Get broker list + brokers := cli.Brokers() + if len(brokers) == 0 { + return errs.New("no brokers found").Wrap() + } + + // Check if all brokers are reachable + for _, broker := range brokers { + if err := broker.Open(kfk); err != nil { + return errs.WrapMsg(err, "failed to open broker", "broker", broker.Addr()) + } + } + + return nil +} diff --git a/pkg/notification/msg.go b/pkg/notification/msg.go index 8e028c577..ba8a9185a 100644 --- a/pkg/notification/msg.go +++ b/pkg/notification/msg.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package rpcclient +package notification import ( "context" @@ -197,7 +197,6 @@ func WithSendMessage(sendMessage *bool) NotificationOptions { } func (s *NotificationSender) send(ctx context.Context, sendID, recvID string, contentType, sessionType int32, m proto.Message, opts ...NotificationOptions) { - //ctx = mcontext.WithMustInfoCtx([]string{mcontext.GetOperationID(ctx), mcontext.GetOpUserID(ctx), mcontext.GetOpUserPlatform(ctx), mcontext.GetConnID(ctx)}) ctx = context.WithoutCancel(ctx) ctx, cancel := context.WithTimeout(ctx, time.Second*time.Duration(5)) defer cancel() @@ -214,7 +213,7 @@ func (s *NotificationSender) send(ctx context.Context, sendID, recvID string, co var req msg.SendMsgReq var msg sdkws.MsgData var userInfo *sdkws.UserInfo - if notificationOpt.WithRpcGetUsername && s.getUserInfo != nil { + if notificationOpt.RpcGetUsername && s.getUserInfo != nil { userInfo, err = s.getUserInfo(ctx, sendID) if err != nil { log.ZWarn(ctx, "getUserInfo failed", err, "sendID", sendID) diff --git a/tools/check-component/main.go b/tools/check-component/main.go index 94dbd613c..cb1fb2a33 100644 --- a/tools/check-component/main.go +++ b/tools/check-component/main.go @@ -26,11 +26,11 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/cmd" "github.com/openimsdk/open-im-server/v3/pkg/common/config" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/kafka" "github.com/openimsdk/tools/db/mongoutil" "github.com/openimsdk/tools/db/redisutil" "github.com/openimsdk/tools/discovery/etcd" "github.com/openimsdk/tools/discovery/zookeeper" - "github.com/openimsdk/tools/mq/kafka" "github.com/openimsdk/tools/s3/minio" "github.com/openimsdk/tools/system/program" ) diff --git a/tools/seq/internal/main.go b/tools/seq/internal/main.go index c9f3a8236..8a4d874bb 100644 --- a/tools/seq/internal/main.go +++ b/tools/seq/internal/main.go @@ -44,7 +44,7 @@ const ( ) func readConfig[T any](dir string, name string) (*T, error) { - if runtimeenv.PrintRuntimeEnvironment() == config.KUBERNETES { + if runtimeenv.RuntimeEnvironment() == config.KUBERNETES { dir = os.Getenv(config.MountConfigFilePath) } v := viper.New()