diff --git a/config/config.yaml b/config/config.yaml
index 2813fed8e..b17d10073 100644
--- a/config/config.yaml
+++ b/config/config.yaml
@@ -158,7 +158,14 @@ object:
accessKeySecret: ''
sessionToken: ''
publicRead: false
-
+ kodo:
+ endpoint: "http://s3.cn-east-1.qiniucs.com"
+ bucket: "demo-9999999"
+ bucketURL: "http://your.domain.com"
+ accessKeyID: ''
+ accessKeySecret: ''
+ sessionToken: ''
+ publicRead: false
###################### RPC Port Configuration ######################
# RPC service ports
# These ports are passed into the program by the script and are not recommended to modify
diff --git a/config/instance-down-rules.yml b/config/instance-down-rules.yml
index 72b1f5aa3..5541d2c54 100644
--- a/config/instance-down-rules.yml
+++ b/config/instance-down-rules.yml
@@ -8,4 +8,15 @@ groups:
severity: critical
annotations:
summary: "Instance {{ $labels.instance }} down"
- description: "{{ $labels.instance }} of job {{ $labels.job }} has been down for more than 1 minutes."
\ No newline at end of file
+ description: "{{ $labels.instance }} of job {{ $labels.job }} has been down for more than 1 minutes."
+
+ - name: database_insert_failure_alerts
+ rules:
+ - alert: DatabaseInsertFailed
+ expr: (increase(msg_insert_redis_failed_total[5m]) > 0) or (increase(msg_insert_mongo_failed_total[5m]) > 0)
+ for: 1m
+ labels:
+ severity: critical
+ annotations:
+ summary: "Increase in MsgInsertRedisFailedCounter or MsgInsertMongoFailedCounter detected"
+ description: "Either MsgInsertRedisFailedCounter or MsgInsertMongoFailedCounter has increased in the last 5 minutes, indicating failures in message insert operations to Redis or MongoDB,maybe the redis or mongodb is crash."
diff --git a/deployments/templates/openim.yaml b/deployments/templates/openim.yaml
index ff2fa896d..fc90bfc2f 100644
--- a/deployments/templates/openim.yaml
+++ b/deployments/templates/openim.yaml
@@ -158,6 +158,14 @@ object:
accessKeySecret: ${OSS_ACCESS_KEY_SECRET}
sessionToken: ${OSS_SESSION_TOKEN}
publicRead: ${OSS_PUBLIC_READ}
+ kodo:
+ endpoint: "${KODO_ENDPOINT}"
+ bucket: "${KODO_BUCKET}"
+ bucketURL: "${KODO_BUCKET_URL}"
+ accessKeyID: ${KODO_ACCESS_KEY_ID}
+ accessKeySecret: ${KODO_ACCESS_KEY_SECRET}
+ sessionToken: ${KODO_SESSION_TOKEN}
+ publicRead: ${KODO_PUBLIC_READ}
###################### RPC Port Configuration ######################
# RPC service ports
diff --git a/docs/contrib/environment.md b/docs/contrib/environment.md
index b82ec0f45..0279c750e 100644
--- a/docs/contrib/environment.md
+++ b/docs/contrib/environment.md
@@ -37,6 +37,7 @@
* 2.20. [Prometheus Configuration](#PrometheusConfiguration-1)
* 2.20.1. [General Configuration](#GeneralConfiguration)
* 2.20.2. [Service-Specific Prometheus Ports](#Service-SpecificPrometheusPorts)
+ * 2.21. [Qiniu Cloud Kodo Configuration](#QiniuCloudKODOConfiguration)
## 0. OpenIM Config File
@@ -528,3 +529,18 @@ This section involves configuring Prometheus, including enabling/disabling it an
| RTC Service | `RTC_PROM_PORT` | '21300' | Prometheus port for the RTC service. |
| Third Service | `THIRD_PROM_PORT` | '21301' | Prometheus port for the Third service. |
| Message Transfer Service | `MSG_TRANSFER_PROM_PORT` | '21400, 21401, 21402, 21403' | Prometheus ports for the Message Transfer service. |
+
+
+### 2.21. Qiniu Cloud Kodo Configuration
+
+This section involves setting up Qiniu Cloud Kodo, including its endpoint, bucket name, and credentials.
+
+| Parameter | Example Value | Description |
+| --------------------- | ------------------------------------------------------------ | ---------------------------------------- |
+| KODO_ENDPOINT | "[http://s3.cn-east-1.qiniucs.com](http://s3.cn-east-1.qiniucs.com)" | Endpoint URL for Qiniu Cloud Kodo. |
+| KODO_BUCKET | "demo-9999999" | Bucket name for Qiniu Cloud Kodo. |
+| KODO_BUCKET_URL | "[http://your.domain.com](http://your.domain.com)" | Bucket URL for Qiniu Cloud Kodo. |
+| KODO_ACCESS_KEY_ID | [User Defined] | Access key ID for Qiniu Cloud Kodo. |
+| KODO_ACCESS_KEY_SECRET | [User Defined] | Access key secret for Qiniu Cloud Kodo. |
+| KODO_SESSION_TOKEN | [User Defined] | Session token for Qiniu Cloud Kodo. |
+| KODO_PUBLIC_READ | "false" | Public read access. |
diff --git a/go.mod b/go.mod
index a5f867d92..545888be7 100644
--- a/go.mod
+++ b/go.mod
@@ -58,6 +58,24 @@ require (
cloud.google.com/go/iam v1.1.2 // indirect
cloud.google.com/go/longrunning v0.5.1 // indirect
cloud.google.com/go/storage v1.30.1 // indirect
+ github.com/aws/aws-sdk-go-v2 v1.23.1 // indirect
+ github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1 // indirect
+ github.com/aws/aws-sdk-go-v2/config v1.25.4 // indirect
+ github.com/aws/aws-sdk-go-v2/credentials v1.16.3 // indirect
+ github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1 // indirect
+ github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4 // indirect
+ github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1 // indirect
+ github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4 // indirect
+ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4 // indirect
+ github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4 // indirect
+ github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1 // indirect
+ github.com/aws/aws-sdk-go-v2/service/sso v1.17.3 // indirect
+ github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1 // indirect
+ github.com/aws/aws-sdk-go-v2/service/sts v1.25.4 // indirect
+ github.com/aws/smithy-go v1.17.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/bytedance/sonic v1.9.1 // indirect
github.com/cespare/xxhash/v2 v2.2.0 // indirect
@@ -115,6 +133,7 @@ require (
github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16 // indirect
github.com/prometheus/common v0.44.0 // indirect
github.com/prometheus/procfs v0.11.1 // indirect
+ github.com/qiniu/go-sdk/v7 v7.18.2 // indirect
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
github.com/rs/xid v1.5.0 // indirect
github.com/sergi/go-diff v1.0.0 // indirect
diff --git a/go.sum b/go.sum
index 0a13abad2..640b11e89 100644
--- a/go.sum
+++ b/go.sum
@@ -31,6 +31,42 @@ github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239 h1:kFOfPq6dUM1hTo
github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
+github.com/aws/aws-sdk-go-v2 v1.23.1 h1:qXaFsOOMA+HsZtX8WoCa+gJnbyW7qyFFBlPqvTSzbaI=
+github.com/aws/aws-sdk-go-v2 v1.23.1/go.mod h1:i1XDttT4rnf6vxc9AuskLc6s7XBee8rlLilKlc03uAA=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1 h1:ZY3108YtBNq96jNZTICHxN1gSBSbnvIdYwwqnvCV4Mc=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1/go.mod h1:t8PYl/6LzdAqsU4/9tz28V/kU+asFePvpOMkdul0gEQ=
+github.com/aws/aws-sdk-go-v2/config v1.25.4 h1:r+X1x8QI6FEPdJDWCNBDZHyAcyFwSjHN8q8uuus+Axs=
+github.com/aws/aws-sdk-go-v2/config v1.25.4/go.mod h1:8GTjImECskr7D88P/Nn9uM4M4rLY9i77hLJZgkZEWV8=
+github.com/aws/aws-sdk-go-v2/credentials v1.16.3 h1:8PeI2krzzjDJ5etmgaMiD1JswsrLrWvKKu/uBUtNy1g=
+github.com/aws/aws-sdk-go-v2/credentials v1.16.3/go.mod h1:Kdh/okh+//vQ/AjEt81CjvkTo64+/zIE4OewP7RpfXk=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5 h1:KehRNiVzIfAcj6gw98zotVbb/K67taJE0fkfgM6vzqU=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.5/go.mod h1:VhnExhw6uXy9QzetvpXDolo1/hjhx4u9qukBGkuUwjs=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4 h1:LAm3Ycm9HJfbSCd5I+wqC2S9Ej7FPrgr5CQoOljJZcE=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.4/go.mod h1:xEhvbJcyUf/31yfGSQBe01fukXwXJ0gxDp7rLfymWE0=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4 h1:4GV0kKZzUxiWxSVpn/9gwR0g21NF1Jsyduzo9rHgC/Q=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.4/go.mod h1:dYvTNAggxDZy6y1AF7YDwXsPuHFy/VNEpEI/2dWK9IU=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1 h1:uR9lXYjdPX0xY+NhvaJ4dD8rpSRz5VY81ccIIoNG+lw=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1/go.mod h1:6fQQgfuGmw8Al/3M2IgIllycxV7ZW7WCdVSqfBeUiCY=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4 h1:40Q4X5ebZruRtknEZH/bg91sT5pR853F7/1X9QRbI54=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.4/go.mod h1:u77N7eEECzUv7F0xl2gcfK/vzc8wcjWobpy+DcrLJ5E=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1 h1:rpkF4n0CyFcrJUG/rNNohoTmhtWlFTRI4BsZOh9PvLs=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1/go.mod h1:l9ymW25HOqymeU2m1gbUQ3rUIsTwKs8gYHXkqDQUhiI=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4 h1:6DRKQc+9cChgzL5gplRGusI5dBGeiEod4m/pmGbcX48=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.4/go.mod h1:s8ORvrW4g4v7IvYKIAoBg17w3GQ+XuwXDXYrQ5SkzU0=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4 h1:rdovz3rEu0vZKbzoMYPTehp0E8veoE9AyfzqCr5Eeao=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.4/go.mod h1:aYCGNjyUCUelhofxlZyj63srdxWUSsBSGg5l6MCuXuE=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4 h1:o3DcfCxGDIT20pTbVKVhp3vWXOj/VvgazNJvumWeYW0=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.4/go.mod h1:Uy0KVOxuTK2ne+/PKQ+VvEeWmjMMksE17k/2RK/r5oM=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1 h1:1w11lfXOa8HoHoSlNtt4mqv/N3HmDOa+OnUH3Y9DHm8=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.43.1/go.mod h1:dqJ5JBL0clzgHriH35Amx3LRFY6wNIPUX7QO/BerSBo=
+github.com/aws/aws-sdk-go-v2/service/sso v1.17.3 h1:CdsSOGlFF3Pn+koXOIpTtvX7st0IuGsZ8kJqcWMlX54=
+github.com/aws/aws-sdk-go-v2/service/sso v1.17.3/go.mod h1:oA6VjNsLll2eVuUoF2D+CMyORgNzPEW/3PyUdq6WQjI=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1 h1:cbRqFTVnJV+KRpwFl76GJdIZJKKCdTPnjUZ7uWh3pIU=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.1/go.mod h1:hHL974p5auvXlZPIjJTblXJpbkfK4klBczlsEaMCGVY=
+github.com/aws/aws-sdk-go-v2/service/sts v1.25.4 h1:yEvZ4neOQ/KpUqyR+X0ycUTW/kVRNR4nDZ38wStHGAA=
+github.com/aws/aws-sdk-go-v2/service/sts v1.25.4/go.mod h1:feTnm2Tk/pJxdX+eooEsxvlvTWBvDm6CasRZ+JOs2IY=
+github.com/aws/smithy-go v1.17.0 h1:wWJD7LX6PBV6etBUwO0zElG0nWN9rUhp0WdYeHSHAaI=
+github.com/aws/smithy-go v1.17.0/go.mod h1:NukqUGpCZIILqqiV0NIjeFh24kd/FAa4beRb6nbIUPE=
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
@@ -56,6 +92,7 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -91,11 +128,17 @@ github.com/gin-gonic/gin v1.9.1 h1:4idEAncQnU5cB7BeOkPtxjfCSye0AAm1R0RVIqJ+Jmg=
github.com/gin-gonic/gin v1.9.1/go.mod h1:hPrL7YrpYKXt5YId3A/Tnip5kqbEAP+KLuI3SUcPTeU=
github.com/gliderlabs/ssh v0.2.2 h1:6zsha5zo/TWhRhwqCD3+EarCAgZ2yN28ipRnGPnwkI0=
github.com/gliderlabs/ssh v0.2.2/go.mod h1:U7qILu1NlMHj9FlMhZLlkCdDnU1DBEAqr0aevW3Awn0=
+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/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
+github.com/go-playground/locales v0.14.0/go.mod h1:sawfccIbzZTqEDETgFXqTho0QybSa7l++s0DH+LDiLs=
github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
+github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
+github.com/go-playground/universal-translator v0.18.0/go.mod h1:UvRDBj+xPUEGrFYl+lu/H90nyDXpg0fqeB/AQUGNTVA=
github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
+github.com/go-playground/validator/v10 v10.8.0/go.mod h1:9JhgTzTaE31GZDpH/HSvHiRJrJ3iKAgqqH0Bl/Ocjdk=
github.com/go-playground/validator/v10 v10.15.5 h1:LEBecTWb/1j5TNY1YYG2RcOUN3R7NLylN+x8TTueE24=
github.com/go-playground/validator/v10 v10.15.5/go.mod h1:9iXMNT7sEkjXb0I+enO7QXmzG6QCsPWY4zveKFVRSyU=
github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg=
@@ -214,11 +257,15 @@ github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa02
github.com/klauspost/cpuid/v2 v2.2.5 h1:0E5MSMDEoAulmXNFquVs//DdoomxaoTY1kUhbc/qbZg=
github.com/klauspost/cpuid/v2 v2.2.5/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/pty v1.1.8/go.mod h1:O1sed60cT9XZ5uDucP5qwvh+TE3NnUj51EiZO/lmSfw=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
github.com/leodido/go-urn v1.2.4 h1:XlAE/cm/ms7TE/VMVoduSpNBoyc2dOxHs5MZSwAN63Q=
github.com/leodido/go-urn v1.2.4/go.mod h1:7ZrI8mTSeBSHl/UaRyKQW1qZeMgak41ANeCNaVckg+4=
github.com/lestrrat-go/envload v0.0.0-20180220234015-a3eb8ddeffcc h1:RKf14vYWi2ttpEmkA4aQ3j4u9dStX2t4M8UM6qqNsG8=
@@ -275,6 +322,7 @@ github.com/pelletier/go-toml/v2 v2.0.8 h1:0ctb6s9mE31h0/lhu+J6OPmVeDxJn+kYnJc2jZ
github.com/pelletier/go-toml/v2 v2.0.8/go.mod h1:vuYfssBdrU2XDZ9bYydBu6t+6a6PYNcZljzZR9VXg+4=
github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -290,12 +338,18 @@ github.com/prometheus/common v0.44.0 h1:+5BrQJwiBB9xsMygAB3TNvpQKOwlkc25LbISbrdO
github.com/prometheus/common v0.44.0/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
github.com/prometheus/procfs v0.11.1 h1:xRC8Iq1yyca5ypa9n1EZnWZkt7dwcoRPQwX/5gwaUuI=
github.com/prometheus/procfs v0.11.1/go.mod h1:eesXgaPo1q7lBpVMoMy0ZOFTth9hBn4W/y0/p/ScXhY=
+github.com/qiniu/dyn v1.3.0/go.mod h1:E8oERcm8TtwJiZvkQPbcAh0RL8jO1G0VXJMW3FAWdkk=
+github.com/qiniu/go-sdk/v7 v7.18.2 h1:vk9eo5OO7aqgAOPF0Ytik/gt7CMKuNgzC/IPkhda6rk=
+github.com/qiniu/go-sdk/v7 v7.18.2/go.mod h1:nqoYCNo53ZlGA521RvRethvxUDvXKt4gtYXOwye868w=
+github.com/qiniu/x v1.10.5/go.mod h1:03Ni9tj+N2h2aKnAz+6N0Xfl8FwMEDRC2PAlxekASDs=
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
github.com/redis/go-redis/v9 v9.2.1 h1:WlYJg71ODF0dVspZZCpYmoF1+U1Jjk9Rwd7pq6QmlCg=
github.com/redis/go-redis/v9 v9.2.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
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.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
github.com/rs/xid v1.5.0 h1:mKX4bl4iPYJtEIxp6CYiUuLQ/8DYMoz0PUdtGgMFRVc=
github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
@@ -318,6 +372,7 @@ github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpE
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
@@ -368,8 +423,10 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.1.0/go.mod h1:RecgLatLF4+eUMCP1PoPZQb+cVrJcOPbHkTkbkB9sbw=
golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
@@ -400,6 +457,7 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v
golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
@@ -432,16 +490,19 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w
golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
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.13.0 h1:bb+I9cTfFazGW51MZqBVmZy7+JEJMouUHTUSKVQLBek=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -450,6 +511,7 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
+golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
@@ -510,6 +572,8 @@ google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqw
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
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/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=
gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
@@ -529,6 +593,7 @@ gopkg.in/yaml.v2 v2.3.0/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=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gorm.io/driver/mysql v1.5.2 h1:QC2HRskSE75wBuOxe0+iCkyJZ+RqpudsQtqkp+IMuXs=
diff --git a/internal/msggateway/init.go b/internal/msggateway/init.go
index 14c320c42..aeba0a24a 100644
--- a/internal/msggateway/init.go
+++ b/internal/msggateway/init.go
@@ -19,6 +19,7 @@ import (
"time"
"github.com/OpenIMSDK/tools/utils"
+ "golang.org/x/sync/errgroup"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
)
@@ -43,12 +44,22 @@ func RunWsAndServer(rpcPort, wsPort, prometheusPort int) error {
if err != nil {
return err
}
+
hubServer := NewServer(rpcPort, prometheusPort, longServer)
- go func() {
- err := hubServer.Start()
+
+ wg := errgroup.Group{}
+ wg.Go(func() error {
+ err = hubServer.Start()
if err != nil {
- panic(utils.Wrap1(err))
+ return utils.Wrap1(err)
}
- }()
- return hubServer.LongConnServer.Run()
+ return err
+ })
+
+ wg.Go(func() error {
+ return hubServer.LongConnServer.Run()
+ })
+
+ err = wg.Wait()
+ return err
}
diff --git a/internal/msggateway/n_ws_server.go b/internal/msggateway/n_ws_server.go
index 5d7d4c542..5a061d7e1 100644
--- a/internal/msggateway/n_ws_server.go
+++ b/internal/msggateway/n_ws_server.go
@@ -18,9 +18,12 @@ import (
"context"
"errors"
"net/http"
+ "os"
+ "os/signal"
"strconv"
"sync"
"sync/atomic"
+ "syscall"
"time"
"github.com/go-playground/validator/v10"
@@ -157,10 +160,22 @@ func NewWsServer(opts ...Option) (*WsServer, error) {
}
func (ws *WsServer) Run() error {
- var client *Client
- go func() {
+ var (
+ client *Client
+ wg errgroup.Group
+
+ sigs = make(chan os.Signal, 1)
+ done = make(chan struct{}, 1)
+ )
+
+ server := http.Server{Addr: ":" + utils.IntToString(ws.port), Handler: nil}
+
+ wg.Go(func() error {
for {
select {
+ case <-done:
+ return nil
+
case client = <-ws.registerChan:
ws.registerClient(client)
case client = <-ws.unregisterChan:
@@ -169,10 +184,34 @@ func (ws *WsServer) Run() error {
ws.multiTerminalLoginChecker(onlineInfo.clientOK, onlineInfo.oldClients, onlineInfo.newClient)
}
}
+ })
+
+ wg.Go(func() error {
+ http.HandleFunc("/", ws.wsHandler)
+ return server.ListenAndServe()
+ })
+
+ signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+ <-sigs
+
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+ defer cancel()
+
+ // graceful exit operation for server
+ _ = server.Shutdown(ctx)
+ _ = wg.Wait()
+ close(done)
}()
- http.HandleFunc("/", ws.wsHandler)
- // http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {})
- return http.ListenAndServe(":"+utils.IntToString(ws.port), nil) // Start listening
+
+ select {
+ case <-done:
+ return nil
+
+ case <-time.After(15 * time.Second):
+ return utils.Wrap1(errors.New("timeout exit"))
+ }
+
}
var concurrentRequest = 3
diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go
index 4ce015543..8436317ee 100644
--- a/internal/msgtransfer/init.go
+++ b/internal/msgtransfer/init.go
@@ -17,16 +17,15 @@ package msgtransfer
import (
"errors"
"fmt"
- "log"
- "net/http"
- "sync"
-
"github.com/OpenIMSDK/tools/mw"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/collectors"
"github.com/prometheus/client_golang/prometheus/promhttp"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
+ "log"
+ "net/http"
+ "sync"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go
index b019b0120..eb8e500fe 100644
--- a/internal/msgtransfer/online_history_msg_handler.go
+++ b/internal/msgtransfer/online_history_msg_handler.go
@@ -252,7 +252,10 @@ func (och *OnlineHistoryRedisConsumerHandler) handleNotification(
return
}
log.ZDebug(ctx, "success to next topic", "conversationID", conversationID)
- och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+ err = och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+ if err != nil {
+ log.ZError(ctx, "MsgToMongoMQ error", err)
+ }
och.toPushTopic(ctx, key, conversationID, storageList)
}
}
@@ -277,9 +280,6 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(
lastSeq, isNewConversation, err := och.msgDatabase.BatchInsertChat2Cache(ctx, conversationID, storageList)
if err != nil && errs.Unwrap(err) != redis.Nil {
log.ZError(ctx, "batch data insert to redis err", err, "storageMsgList", storageList)
- och.singleMsgFailedCountMutex.Lock()
- och.singleMsgFailedCount += uint64(len(storageList))
- och.singleMsgFailedCountMutex.Unlock()
return
}
if isNewConversation {
@@ -311,10 +311,10 @@ func (och *OnlineHistoryRedisConsumerHandler) handleMsg(
}
log.ZDebug(ctx, "success incr to next topic")
- och.singleMsgSuccessCountMutex.Lock()
- och.singleMsgSuccessCount += uint64(len(storageList))
- och.singleMsgSuccessCountMutex.Unlock()
- och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+ err = och.msgDatabase.MsgToMongoMQ(ctx, key, conversationID, storageList, lastSeq)
+ if err != nil {
+ log.ZError(ctx, "MsgToMongoMQ error", err)
+ }
och.toPushTopic(ctx, key, conversationID, storageList)
}
}
diff --git a/internal/rpc/msg/sync_msg.go b/internal/rpc/msg/sync_msg.go
index 7c67ff05f..dbd8da4d8 100644
--- a/internal/rpc/msg/sync_msg.go
+++ b/internal/rpc/msg/sync_msg.go
@@ -42,15 +42,8 @@ func (m *msgServer) PullMessageBySeqs(
log.ZError(ctx, "GetConversation error", err, "conversationID", seq.ConversationID)
continue
}
- minSeq, maxSeq, msgs, err := m.MsgDatabase.GetMsgBySeqsRange(
- ctx,
- req.UserID,
- seq.ConversationID,
- seq.Begin,
- seq.End,
- seq.Num,
- conversation.MaxSeq,
- )
+ minSeq, maxSeq, msgs, err := m.MsgDatabase.GetMsgBySeqsRange(ctx, req.UserID, seq.ConversationID,
+ seq.Begin, seq.End, seq.Num, conversation.MaxSeq)
if err != nil {
log.ZWarn(ctx, "GetMsgBySeqsRange error", err, "conversationID", seq.ConversationID, "seq", seq)
continue
@@ -64,7 +57,6 @@ func (m *msgServer) PullMessageBySeqs(
}
if len(msgs) == 0 {
log.ZWarn(ctx, "not have msgs", nil, "conversationID", seq.ConversationID, "seq", seq)
-
continue
}
resp.Msgs[seq.ConversationID] = &sdkws.PullMsgs{Msgs: msgs, IsEnd: isEnd}
diff --git a/internal/rpc/third/third.go b/internal/rpc/third/third.go
index b48eddea9..eed3d4802 100644
--- a/internal/rpc/third/third.go
+++ b/internal/rpc/third/third.go
@@ -17,14 +17,15 @@ package third
import (
"context"
"fmt"
+
"net/url"
"time"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/cos"
+ "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/kodo"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/minio"
"github.com/openimsdk/open-im-server/v3/pkg/common/db/s3/oss"
-
"google.golang.org/grpc"
"github.com/OpenIMSDK/protocol/third"
@@ -72,6 +73,8 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e
o, err = cos.NewCos()
case "oss":
o, err = oss.NewOSS()
+ case "kodo":
+ o, err = kodo.NewKodo()
default:
err = fmt.Errorf("invalid object enable: %s", enable)
}
diff --git a/pkg/common/cmd/msg_gateway.go b/pkg/common/cmd/msg_gateway.go
index 7f0abb771..25fcc1177 100644
--- a/pkg/common/cmd/msg_gateway.go
+++ b/pkg/common/cmd/msg_gateway.go
@@ -17,12 +17,12 @@ package cmd
import (
"log"
- "github.com/openimsdk/open-im-server/v3/internal/msggateway"
- v3config "github.com/openimsdk/open-im-server/v3/pkg/common/config"
-
"github.com/spf13/cobra"
"github.com/OpenIMSDK/protocol/constant"
+
+ "github.com/openimsdk/open-im-server/v3/internal/msggateway"
+ v3config "github.com/openimsdk/open-im-server/v3/pkg/common/config"
)
type MsgGatewayCmd struct {
@@ -60,14 +60,19 @@ func (m *MsgGatewayCmd) Exec() error {
m.addRunE()
return m.Execute()
}
+
func (m *MsgGatewayCmd) GetPortFromConfig(portType string) int {
- if portType == constant.FlagWsPort {
+ switch portType {
+ case constant.FlagWsPort:
return v3config.Config.LongConnSvr.OpenImWsPort[0]
- } else if portType == constant.FlagPort {
+
+ case constant.FlagPort:
return v3config.Config.LongConnSvr.OpenImMessageGatewayPort[0]
- } else if portType == constant.FlagPrometheusPort {
+
+ case constant.FlagPrometheusPort:
return v3config.Config.Prometheus.MessageGatewayPrometheusPort[0]
- } else {
+
+ default:
return 0
}
}
diff --git a/pkg/common/config/config.go b/pkg/common/config/config.go
index 65258d202..3c819ea2c 100644
--- a/pkg/common/config/config.go
+++ b/pkg/common/config/config.go
@@ -152,6 +152,15 @@ type configStruct struct {
SessionToken string `yaml:"sessionToken"`
PublicRead bool `yaml:"publicRead"`
} `yaml:"oss"`
+ Kodo struct {
+ Endpoint string `yaml:"endpoint"`
+ Bucket string `yaml:"bucket"`
+ BucketURL string `yaml:"bucketURL"`
+ AccessKeyID string `yaml:"accessKeyID"`
+ AccessKeySecret string `yaml:"accessKeySecret"`
+ SessionToken string `yaml:"sessionToken"`
+ PublicRead bool `yaml:"publicRead"`
+ } `yaml:"kodo"`
} `yaml:"object"`
RpcPort struct {
diff --git a/pkg/common/db/cache/msg.go b/pkg/common/db/cache/msg.go
index 282d1d1c1..f86b44d9b 100644
--- a/pkg/common/db/cache/msg.go
+++ b/pkg/common/db/cache/msg.go
@@ -173,7 +173,20 @@ func (c *msgCache) getSeqs(ctx context.Context, items []string, getkey func(s st
}
func (c *msgCache) SetMaxSeq(ctx context.Context, conversationID string, maxSeq int64) error {
- return c.setSeq(ctx, conversationID, maxSeq, c.getMaxSeqKey)
+ var retErr error
+ for {
+ select {
+ case <-ctx.Done():
+ return errs.Wrap(retErr, "SetMaxSeq redis retry too many amount")
+ default:
+ retErr = c.setSeq(ctx, conversationID, maxSeq, c.getMaxSeqKey)
+ if retErr != nil {
+ time.Sleep(time.Second * 2)
+ continue
+ }
+ return nil
+ }
+ }
}
func (c *msgCache) GetMaxSeqs(ctx context.Context, conversationIDs []string) (m map[string]int64, err error) {
@@ -181,7 +194,21 @@ func (c *msgCache) GetMaxSeqs(ctx context.Context, conversationIDs []string) (m
}
func (c *msgCache) GetMaxSeq(ctx context.Context, conversationID string) (int64, error) {
- return c.getSeq(ctx, conversationID, c.getMaxSeqKey)
+ var retErr error
+ var retData int64
+ for {
+ select {
+ case <-ctx.Done():
+ return -1, errs.Wrap(retErr, "GetMaxSeq redis retry too many amount")
+ default:
+ retData, retErr = c.getSeq(ctx, conversationID, c.getMaxSeqKey)
+ if retErr != nil && errs.Unwrap(retErr) != redis.Nil {
+ time.Sleep(time.Second * 2)
+ continue
+ }
+ return retData, retErr
+ }
+ }
}
func (c *msgCache) SetMinSeq(ctx context.Context, conversationID string, minSeq int64) error {
diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go
index fb0a9c702..cba0a6bbd 100644
--- a/pkg/common/db/controller/msg.go
+++ b/pkg/common/db/controller/msg.go
@@ -357,7 +357,9 @@ func (db *commonMsgDatabase) DelUserDeleteMsgsList(ctx context.Context, conversa
}
func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversationID string, msgs []*sdkws.MsgData) (seq int64, isNew bool, err error) {
- currentMaxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
+ cancelCtx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+ defer cancel()
+ currentMaxSeq, err := db.cache.GetMaxSeq(cancelCtx, conversationID)
if err != nil && errs.Unwrap(err) != redis.Nil {
log.ZError(ctx, "db.cache.GetMaxSeq", err)
return 0, false, err
@@ -384,19 +386,21 @@ func (db *commonMsgDatabase) BatchInsertChat2Cache(ctx context.Context, conversa
prommetrics.MsgInsertRedisFailedCounter.Add(float64(failedNum))
log.ZError(ctx, "setMessageToCache error", err, "len", len(msgs), "conversationID", conversationID)
} else {
- prommetrics.MsgInsertRedisSuccessCounter.Inc()
+ prommetrics.MsgInsertRedisSuccessCounter.Add(float64(len(msgs)))
}
- err = db.cache.SetMaxSeq(ctx, conversationID, currentMaxSeq)
+ cancelCtx, cancel = context.WithTimeout(ctx, 1*time.Minute)
+ defer cancel()
+ err = db.cache.SetMaxSeq(cancelCtx, conversationID, currentMaxSeq)
if err != nil {
log.ZError(ctx, "db.cache.SetMaxSeq error", err, "conversationID", conversationID)
prommetrics.SeqSetFailedCounter.Inc()
}
err2 := db.cache.SetHasReadSeqs(ctx, conversationID, userSeqMap)
- if err != nil {
+ if err2 != nil {
log.ZError(ctx, "SetHasReadSeqs error", err2, "userSeqMap", userSeqMap, "conversationID", conversationID)
prommetrics.SeqSetFailedCounter.Inc()
}
- return lastMaxSeq, isNew, utils.Wrap(err, "")
+ return lastMaxSeq, isNew, errs.Wrap(err, "redis SetMaxSeq error")
}
func (db *commonMsgDatabase) getMsgBySeqs(ctx context.Context, userID, conversationID string, seqs []int64) (totalMsgs []*sdkws.MsgData, err error) {
@@ -654,16 +658,26 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin
func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) (int64, int64, []*sdkws.MsgData, error) {
userMinSeq, err := db.cache.GetConversationUserMinSeq(ctx, conversationID, userID)
- if err != nil && errs.Unwrap(err) != redis.Nil {
- return 0, 0, nil, err
+ if err != nil {
+ log.ZError(ctx, "cache.GetConversationUserMinSeq error", err)
+ if errs.Unwrap(err) != redis.Nil {
+ return 0, 0, nil, err
+ }
}
minSeq, err := db.cache.GetMinSeq(ctx, conversationID)
- if err != nil && errs.Unwrap(err) != redis.Nil {
- return 0, 0, nil, err
+ if err != nil {
+ log.ZError(ctx, "cache.GetMinSeq error", err)
+ if errs.Unwrap(err) != redis.Nil {
+ return 0, 0, nil, err
+ }
}
maxSeq, err := db.cache.GetMaxSeq(ctx, conversationID)
- if err != nil && errs.Unwrap(err) != redis.Nil {
- return 0, 0, nil, err
+ if err != nil {
+ log.ZError(ctx, "cache.GetMaxSeq error", err)
+ if errs.Unwrap(err) != redis.Nil {
+ return 0, 0, nil, err
+ }
+
}
if userMinSeq < minSeq {
minSeq = userMinSeq
@@ -676,34 +690,16 @@ func (db *commonMsgDatabase) GetMsgBySeqs(ctx context.Context, userID string, co
}
successMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, conversationID, newSeqs)
if err != nil {
- if err != redis.Nil {
- log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
- }
+ log.ZError(ctx, "get message from redis exception", err, "failedSeqs", failedSeqs, "conversationID", conversationID)
}
- log.ZInfo(
- ctx,
- "db.cache.GetMessagesBySeq",
- "userID",
- userID,
- "conversationID",
- conversationID,
- "seqs",
- seqs,
- "successMsgs",
- len(successMsgs),
- "failedSeqs",
- failedSeqs,
- "conversationID",
- conversationID,
- )
+ log.ZInfo(ctx, "db.cache.GetMessagesBySeq", "userID", userID, "conversationID", conversationID, "seqs", seqs, "successMsgs",
+ len(successMsgs), "failedSeqs", failedSeqs, "conversationID", conversationID)
if len(failedSeqs) > 0 {
mongoMsgs, err := db.getMsgBySeqs(ctx, userID, conversationID, failedSeqs)
if err != nil {
-
return 0, 0, nil, err
}
-
successMsgs = append(successMsgs, mongoMsgs...)
}
return minSeq, maxSeq, successMsgs, nil
diff --git a/pkg/common/db/s3/kodo/internal.go b/pkg/common/db/s3/kodo/internal.go
new file mode 100644
index 000000000..3a4943e62
--- /dev/null
+++ b/pkg/common/db/s3/kodo/internal.go
@@ -0,0 +1 @@
+package kodo
diff --git a/pkg/common/db/s3/kodo/kodo.go b/pkg/common/db/s3/kodo/kodo.go
new file mode 100644
index 000000000..d73220b3b
--- /dev/null
+++ b/pkg/common/db/s3/kodo/kodo.go
@@ -0,0 +1,323 @@
+package kodo
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net/http"
+ "net/url"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awss3config "github.com/aws/aws-sdk-go-v2/config"
+ "github.com/aws/aws-sdk-go-v2/credentials"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ awss3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+ "github.com/openimsdk/open-im-server/v3/pkg/common/config"
+ "github.com/openimsdk/open-im-server/v3/pkg/common/db/s3"
+ "github.com/qiniu/go-sdk/v7/auth"
+ "github.com/qiniu/go-sdk/v7/storage"
+)
+
+const (
+ minPartSize = 1024 * 1024 * 1 // 1MB
+ maxPartSize = 1024 * 1024 * 1024 * 5 // 5GB
+ maxNumSize = 10000
+)
+
+type Kodo struct {
+ AccessKey string
+ SecretKey string
+ Region string
+ Token string
+ Endpoint string
+ BucketURL string
+ Auth *auth.Credentials
+ Client *awss3.Client
+ PresignClient *awss3.PresignClient
+}
+
+func NewKodo() (s3.Interface, error) {
+ conf := config.Config.Object.Kodo
+ //init client
+ cfg, err := awss3config.LoadDefaultConfig(context.TODO(),
+ awss3config.WithRegion(conf.Bucket),
+ awss3config.WithEndpointResolverWithOptions(
+ aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+ return aws.Endpoint{URL: conf.Endpoint}, nil
+ })),
+ awss3config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
+ conf.AccessKeyID,
+ conf.AccessKeySecret,
+ conf.SessionToken),
+ ),
+ )
+ if err != nil {
+ panic(err)
+ }
+ client := awss3.NewFromConfig(cfg)
+ presignClient := awss3.NewPresignClient(client)
+
+ return &Kodo{
+ AccessKey: conf.AccessKeyID,
+ SecretKey: conf.AccessKeySecret,
+ Region: conf.Bucket,
+ BucketURL: conf.BucketURL,
+ Auth: auth.New(conf.AccessKeyID, conf.AccessKeySecret),
+ Client: client,
+ PresignClient: presignClient,
+ }, nil
+}
+
+func (k Kodo) Engine() string {
+ return "kodo"
+}
+
+func (k Kodo) PartLimit() *s3.PartLimit {
+ return &s3.PartLimit{
+ MinPartSize: minPartSize,
+ MaxPartSize: maxPartSize,
+ MaxNumSize: maxNumSize,
+ }
+}
+
+func (k Kodo) InitiateMultipartUpload(ctx context.Context, name string) (*s3.InitiateMultipartUploadResult, error) {
+ result, err := k.Client.CreateMultipartUpload(ctx, &awss3.CreateMultipartUploadInput{
+ Bucket: aws.String(k.Region),
+ Key: aws.String(name),
+ })
+ if err != nil {
+ return nil, err
+ }
+ return &s3.InitiateMultipartUploadResult{
+ UploadID: aws.ToString(result.UploadId),
+ Bucket: aws.ToString(result.Bucket),
+ Key: aws.ToString(result.Key),
+ }, nil
+}
+
+func (k Kodo) CompleteMultipartUpload(ctx context.Context, uploadID string, name string, parts []s3.Part) (*s3.CompleteMultipartUploadResult, error) {
+ kodoParts := make([]awss3types.CompletedPart, len(parts))
+ for i, part := range parts {
+ kodoParts[i] = awss3types.CompletedPart{
+ PartNumber: aws.Int32(int32(part.PartNumber)),
+ ETag: aws.String(part.ETag),
+ }
+ }
+ result, err := k.Client.CompleteMultipartUpload(ctx, &awss3.CompleteMultipartUploadInput{
+ Bucket: aws.String(k.Region),
+ Key: aws.String(name),
+ UploadId: aws.String(uploadID),
+ MultipartUpload: &awss3types.CompletedMultipartUpload{Parts: kodoParts},
+ })
+ if err != nil {
+ return nil, err
+ }
+ return &s3.CompleteMultipartUploadResult{
+ Location: aws.ToString(result.Location),
+ Bucket: aws.ToString(result.Bucket),
+ Key: aws.ToString(result.Key),
+ ETag: strings.ToLower(strings.ReplaceAll(aws.ToString(result.ETag), `"`, ``)),
+ }, nil
+}
+
+func (k Kodo) PartSize(ctx context.Context, size int64) (int64, error) {
+ if size <= 0 {
+ return 0, errors.New("size must be greater than 0")
+ }
+ if size > maxPartSize*maxNumSize {
+ return 0, fmt.Errorf("size must be less than %db", maxPartSize*maxNumSize)
+ }
+ if size <= minPartSize*maxNumSize {
+ return minPartSize, nil
+ }
+ partSize := size / maxNumSize
+ if size%maxNumSize != 0 {
+ partSize++
+ }
+ return partSize, nil
+}
+
+func (k Kodo) AuthSign(ctx context.Context, uploadID string, name string, expire time.Duration, partNumbers []int) (*s3.AuthSignResult, error) {
+ result := s3.AuthSignResult{
+ URL: k.BucketURL + "/" + name,
+ Query: url.Values{"uploadId": {uploadID}},
+ Header: make(http.Header),
+ Parts: make([]s3.SignPart, len(partNumbers)),
+ }
+ for i, partNumber := range partNumbers {
+ part, _ := k.PresignClient.PresignUploadPart(ctx, &awss3.UploadPartInput{
+ Bucket: aws.String(k.Region),
+ UploadId: aws.String(uploadID),
+ Key: aws.String(name),
+ PartNumber: aws.Int32(int32(partNumber)),
+ })
+ result.Parts[i] = s3.SignPart{
+ PartNumber: partNumber,
+ URL: part.URL,
+ Header: part.SignedHeader,
+ }
+ }
+ return &result, nil
+
+}
+
+func (k Kodo) PresignedPutObject(ctx context.Context, name string, expire time.Duration) (string, error) {
+ object, err := k.PresignClient.PresignPutObject(ctx, &awss3.PutObjectInput{
+ Bucket: aws.String(k.Region),
+ Key: aws.String(name),
+ }, func(po *awss3.PresignOptions) {
+ po.Expires = expire
+ })
+ return object.URL, err
+
+}
+
+func (k Kodo) DeleteObject(ctx context.Context, name string) error {
+ _, err := k.Client.DeleteObject(ctx, &awss3.DeleteObjectInput{
+ Bucket: aws.String(k.Region),
+ Key: aws.String(name),
+ })
+ return err
+}
+
+func (k Kodo) CopyObject(ctx context.Context, src string, dst string) (*s3.CopyObjectInfo, error) {
+ result, err := k.Client.CopyObject(ctx, &awss3.CopyObjectInput{
+ Bucket: aws.String(k.Region),
+ CopySource: aws.String(k.Region + "/" + src),
+ Key: aws.String(dst),
+ })
+ if err != nil {
+ return nil, err
+ }
+ return &s3.CopyObjectInfo{
+ Key: dst,
+ ETag: strings.ToLower(strings.ReplaceAll(aws.ToString(result.CopyObjectResult.ETag), `"`, ``)),
+ }, nil
+}
+
+func (k Kodo) StatObject(ctx context.Context, name string) (*s3.ObjectInfo, error) {
+ info, err := k.Client.HeadObject(ctx, &awss3.HeadObjectInput{
+ Bucket: aws.String(k.Region),
+ Key: aws.String(name),
+ })
+ if err != nil {
+ return nil, err
+ }
+ res := &s3.ObjectInfo{Key: name}
+ res.Size = aws.ToInt64(info.ContentLength)
+ res.ETag = strings.ToLower(strings.ReplaceAll(aws.ToString(info.ETag), `"`, ``))
+ return res, nil
+}
+
+func (k Kodo) IsNotFound(err error) bool {
+ return true
+}
+
+func (k Kodo) AbortMultipartUpload(ctx context.Context, uploadID string, name string) error {
+ _, err := k.Client.AbortMultipartUpload(ctx, &awss3.AbortMultipartUploadInput{
+ UploadId: aws.String(uploadID),
+ Bucket: aws.String(k.Region),
+ Key: aws.String(name),
+ })
+ return err
+}
+
+func (k Kodo) ListUploadedParts(ctx context.Context, uploadID string, name string, partNumberMarker int, maxParts int) (*s3.ListUploadedPartsResult, error) {
+ result, err := k.Client.ListParts(ctx, &awss3.ListPartsInput{
+ Key: aws.String(name),
+ UploadId: aws.String(uploadID),
+ Bucket: aws.String(k.Region),
+ MaxParts: aws.Int32(int32(maxParts)),
+ PartNumberMarker: aws.String(strconv.Itoa(partNumberMarker)),
+ })
+ if err != nil {
+ return nil, err
+ }
+ res := &s3.ListUploadedPartsResult{
+ Key: aws.ToString(result.Key),
+ UploadID: aws.ToString(result.UploadId),
+ MaxParts: int(aws.ToInt32(result.MaxParts)),
+ UploadedParts: make([]s3.UploadedPart, len(result.Parts)),
+ }
+ // int to string
+ NextPartNumberMarker, err := strconv.Atoi(aws.ToString(result.NextPartNumberMarker))
+ if err != nil {
+ return nil, err
+ }
+ res.NextPartNumberMarker = NextPartNumberMarker
+ for i, part := range result.Parts {
+ res.UploadedParts[i] = s3.UploadedPart{
+ PartNumber: int(aws.ToInt32(part.PartNumber)),
+ LastModified: aws.ToTime(part.LastModified),
+ ETag: aws.ToString(part.ETag),
+ Size: aws.ToInt64(part.Size),
+ }
+ }
+ return res, nil
+}
+
+func (k Kodo) AccessURL(ctx context.Context, name string, expire time.Duration, opt *s3.AccessURLOption) (string, error) {
+ //get object head
+ info, err := k.Client.HeadObject(ctx, &awss3.HeadObjectInput{
+ Bucket: aws.String(k.Region),
+ Key: aws.String(name),
+ })
+ if err != nil {
+ return "", errors.New("AccessURL object not found")
+ }
+ if opt != nil {
+ if opt.ContentType != aws.ToString(info.ContentType) {
+ //修改文件类型
+ err := k.SetObjectContentType(ctx, name, opt.ContentType)
+ if err != nil {
+ return "", errors.New("AccessURL setContentType error")
+ }
+ }
+ }
+ imageMogr := ""
+ //image dispose
+ if opt != nil {
+ if opt.Image != nil {
+ //https://developer.qiniu.com/dora/8255/the-zoom
+ process := ""
+ if opt.Image.Width > 0 {
+ process += strconv.Itoa(opt.Image.Width) + "x"
+ }
+ if opt.Image.Height > 0 {
+ if opt.Image.Width > 0 {
+ process += strconv.Itoa(opt.Image.Height)
+ } else {
+ process += "x" + strconv.Itoa(opt.Image.Height)
+ }
+ }
+ imageMogr = "imageMogr2/thumbnail/" + process
+ }
+ }
+ //expire
+ deadline := time.Now().Add(time.Second * expire).Unix()
+ domain := k.BucketURL
+ query := url.Values{}
+ if opt != nil && opt.Filename != "" {
+ query.Add("attname", opt.Filename)
+ }
+ privateURL := storage.MakePrivateURLv2WithQuery(k.Auth, domain, name, query, deadline)
+ if imageMogr != "" {
+ privateURL += "&" + imageMogr
+ }
+ return privateURL, nil
+}
+
+func (k *Kodo) SetObjectContentType(ctx context.Context, name string, contentType string) error {
+ //set object content-type
+ _, err := k.Client.CopyObject(ctx, &awss3.CopyObjectInput{
+ Bucket: aws.String(k.Region),
+ CopySource: aws.String(k.Region + "/" + name),
+ Key: aws.String(name),
+ ContentType: aws.String(contentType),
+ MetadataDirective: awss3types.MetadataDirectiveReplace,
+ })
+ return err
+}
diff --git a/pkg/common/startrpc/start.go b/pkg/common/startrpc/start.go
index d5e31701e..01076bbbb 100644
--- a/pkg/common/startrpc/start.go
+++ b/pkg/common/startrpc/start.go
@@ -15,14 +15,21 @@
package startrpc
import (
+ "errors"
"fmt"
"log"
"net"
"net/http"
+ "os"
+ "os/signal"
"strconv"
+ "sync"
+ "syscall"
+ "time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
+ "golang.org/x/sync/errgroup"
"github.com/openimsdk/open-im-server/v3/pkg/common/config"
"github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics"
@@ -56,31 +63,37 @@ func Start(
if err != nil {
return err
}
+
defer listener.Close()
client, err := kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery)
if err != nil {
return utils.Wrap1(err)
}
+
defer client.Close()
client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()))
registerIP, err := network.GetRpcRegisterIP(config.Config.Rpc.RegisterIP)
if err != nil {
return err
}
+
var reg *prometheus.Registry
var metric *grpcprometheus.ServerMetrics
- // ctx 中间件
if config.Config.Prometheus.Enable {
- //////////////////////////
cusMetrics := prommetrics.GetGrpcCusMetrics(rpcRegisterName)
- reg, metric, err = prommetrics.NewGrpcPromObj(cusMetrics)
+ reg, metric, _ = prommetrics.NewGrpcPromObj(cusMetrics)
options = append(options, mw.GrpcServer(), grpc.StreamInterceptor(metric.StreamServerInterceptor()),
grpc.UnaryInterceptor(metric.UnaryServerInterceptor()))
} else {
options = append(options, mw.GrpcServer())
}
+
srv := grpc.NewServer(options...)
- defer srv.GracefulStop()
+ once := sync.Once{}
+ defer func() {
+ once.Do(srv.GracefulStop)
+ }()
+
err = rpcFn(client, srv)
if err != nil {
return utils.Wrap1(err)
@@ -94,7 +107,10 @@ func Start(
if err != nil {
return utils.Wrap1(err)
}
- go func() {
+
+ var wg errgroup.Group
+
+ wg.Go(func() error {
if config.Config.Prometheus.Enable && prometheusPort != 0 {
metric.InitializeMetrics(srv)
// Create a HTTP server for prometheus.
@@ -103,7 +119,34 @@ func Start(
log.Fatal("Unable to start a http server.")
}
}
+ return nil
+ })
+
+ wg.Go(func() error {
+ return utils.Wrap1(srv.Serve(listener))
+ })
+
+ sigs := make(chan os.Signal, 1)
+ signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+ <-sigs
+
+ var (
+ done = make(chan struct{}, 1)
+ gerr error
+ )
+
+ go func() {
+ once.Do(srv.GracefulStop)
+ gerr = wg.Wait()
+ close(done)
}()
- return utils.Wrap1(srv.Serve(listener))
+ select {
+ case <-done:
+ return gerr
+
+ case <-time.After(15 * time.Second):
+ return utils.Wrap1(errors.New("timeout exit"))
+ }
+
}
diff --git a/scripts/install/environment.sh b/scripts/install/environment.sh
index b0fd1b65c..b7c8a5dcb 100755
--- a/scripts/install/environment.sh
+++ b/scripts/install/environment.sh
@@ -223,6 +223,15 @@ def "OSS_ACCESS_KEY_SECRET" # 阿里
def "OSS_SESSION_TOKEN" # 阿里云OSS的会话令牌
def "OSS_PUBLIC_READ" "false" # 公有读
+#七牛云配置信息
+def "KODO_ENDPOINT" "http://s3.cn-east-1.qiniucs.com" # 七牛云OSS的端点URL
+def "KODO_BUCKET" "demo-9999999" # 七牛云OSS的存储桶名称
+def "KODO_BUCKET_URL" "http://your.domain.com" # 七牛云OSS的存储桶URL
+def "KODO_ACCESS_KEY_ID" # 七牛云OSS的访问密钥ID
+def "KODO_ACCESS_KEY_SECRET" # 七牛云OSS的密钥
+def "KODO_SESSION_TOKEN" # 七牛云OSS的会话令牌
+def "KODO_PUBLIC_READ" "false" # 公有读
+
###################### Redis 配置信息 ######################
def "REDIS_PORT" "16379" # Redis的端口
def "REDIS_ADDRESS" "${DOCKER_BRIDGE_GATEWAY}" # Redis的地址