diff --git a/common/constant/key.go b/common/constant/key.go index 12d41c794c..d7fb47884c 100644 --- a/common/constant/key.go +++ b/common/constant/key.go @@ -267,8 +267,6 @@ const ( const ( PolarisKey = "polaris" PolarisDefaultRoleType = 3 - PolarisConfigFilePath = "configPath" - PolarisNamespace = "namespace" PolarisServiceToken = "token" PolarisServiceNameSeparator = ":" PolarisDubboPath = "DUBBOPATH" diff --git a/go.mod b/go.mod index 27701f76b4..b67c712bc4 100644 --- a/go.mod +++ b/go.mod @@ -33,13 +33,14 @@ require ( github.com/jinzhu/copier v0.3.5 github.com/knadh/koanf v1.4.4 github.com/magiconair/properties v1.8.6 + github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect github.com/mitchellh/mapstructure v1.5.0 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd github.com/nacos-group/nacos-sdk-go/v2 v2.1.2 github.com/natefinch/lumberjack v2.0.0+incompatible github.com/opentracing/opentracing-go v1.2.0 github.com/pkg/errors v0.9.1 - github.com/polarismesh/polaris-go v1.1.0 + github.com/polarismesh/polaris-go v1.2.0 github.com/prometheus/client_golang v1.12.2 github.com/satori/go.uuid v1.2.1-0.20181028125025-b2ce2384e17b github.com/stretchr/testify v1.8.1 @@ -48,8 +49,9 @@ require ( go.opentelemetry.io/otel v1.11.0 go.opentelemetry.io/otel/trace v1.11.0 go.uber.org/atomic v1.9.0 + go.uber.org/multierr v1.8.0 // indirect go.uber.org/zap v1.21.0 - google.golang.org/genproto v0.0.0-20211104193956-4c6863e31247 + google.golang.org/genproto v0.0.0-20220504150022-98cd25cafc72 google.golang.org/grpc v1.50.1 google.golang.org/protobuf v1.28.1 gopkg.in/yaml.v2 v2.4.0 diff --git a/go.sum b/go.sum index 5d03ba488a..47f43871e1 100644 --- a/go.sum +++ b/go.sum @@ -168,6 +168,8 @@ 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= github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no= +github.com/dlclark/regexp2 v1.7.0 h1:7lJfhqlPssTb1WQx4yvTHN0uElPEv52sbaECrAQxjAo= +github.com/dlclark/regexp2 v1.7.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/dubbogo/go-zookeeper v1.0.3/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c= github.com/dubbogo/go-zookeeper v1.0.4-0.20211212162352-f9d2183d89d5 h1:XoR8SSVziXe698dt4uZYDfsmHpKLemqAgFyndQsq5Kw= github.com/dubbogo/go-zookeeper v1.0.4-0.20211212162352-f9d2183d89d5/go.mod h1:fn6n2CAEer3novYgk9ULLwAjuV8/g4DdC2ENwRb6E+c= @@ -226,7 +228,6 @@ github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw= github.com/getsentry/raven-go v0.2.0 h1:no+xWJRb5ZI7eE8TWgIq1jLulQiIoLG0IfYxv5JYMGs= github.com/getsentry/raven-go v0.2.0/go.mod h1:KungGk8q33+aIAZUIVWZDr2OfAEBsO49PX4NzFV5kcQ= -github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk= github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04= github.com/go-asn1-ber/asn1-ber v1.3.1/go.mod h1:hEBeB/ic+5LoWskz+yKT7vGhhPYkProFKoKdwZRWMe0= github.com/go-co-op/gocron v1.9.0 h1:+V+DDenw3ryB7B+tK1bAIC5p0ruw4oX9IqAsdRnGIf0= @@ -539,8 +540,9 @@ github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Ky github.com/mattn/go-isatty v0.0.14 h1:yVuAays6BHfxijgZPzw+3Zlu5yQgKGP2/hcQbHb7S9Y= github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94= github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU= -github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= +github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo= +github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4= github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= github.com/miekg/dns v1.1.26/go.mod h1:bPDLeHnStXmXAq1m/Ch/hvfNHr14JKNPMBo3VZKjuso= github.com/miekg/dns v1.1.41/go.mod h1:p6aan82bvRIyn+zDIv9xYNUpwa73JcSh9BKwknJysuI= @@ -630,8 +632,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/polarismesh/polaris-go v1.1.0 h1:nFvn3q3XaVFhzF7pBnIySrN0ZZBwvbbYXC5r2DpsQN0= -github.com/polarismesh/polaris-go v1.1.0/go.mod h1:tquawfjEKp1W3ffNJQSzhfditjjoZ7tvhOCElN7Efzs= +github.com/polarismesh/polaris-go v1.2.0 h1:GIWvUno4DB0PD3HC7bHmTCeG+a54neMrVvNPiRXJUtc= +github.com/polarismesh/polaris-go v1.2.0/go.mod h1:HsN0ierETIujHpmnnYJ3qkwQw4QGAECuHvBZTDaw1tI= github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI= github.com/posener/complete v1.2.3/go.mod h1:WZIdtGGp+qx0sLrYKtIRAruyNpv6hFCicSgv7Sy7s/s= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= @@ -833,21 +835,20 @@ go.uber.org/atomic v1.6.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= 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/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= 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/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4= go.uber.org/multierr v1.5.0/go.mod h1:FeouvMocqHpRaaGuG9EjoKcStLC43Zu/fmqdUMPcKYU= -go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4= go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +go.uber.org/multierr v1.8.0 h1:dg6GjLku4EH+249NNmoIciG9N/jURbDG+pFlTkhzIC8= +go.uber.org/multierr v1.8.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak= go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA= go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= go.uber.org/zap v1.13.0/go.mod h1:zwrFLgMcdUuIBviXEYEH1YKNaOBnKXsx2IPda5bBwHM= go.uber.org/zap v1.15.0/go.mod h1:Mb2vm2krFEG5DV0W9qcHBYFtp/Wku1cvYaqPsS/WYfc= go.uber.org/zap v1.16.0/go.mod h1:MA8QOfq0BHJwdXa996Y4dYkAqRKB8/1K1QMMZVaNZjQ= go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo= -go.uber.org/zap v1.19.1/go.mod h1:j3DNczoxDZroyBnOT1L/Q79cfUMGZxlv/9dzN7SM1rI= go.uber.org/zap v1.21.0 h1:WefMeulhovoZ2sYXz7st6K0sLj7bBhpiFaud4r4zST8= go.uber.org/zap v1.21.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= @@ -948,11 +949,11 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210410081132-afb366fc7cd1/go.mod h1:9tjilg8BloeKEkVJvy7fQ90B1CfIiPueXVOjqfkSzI8= golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.0.0-20210917221730-978cfadd31cf/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211029224645-99673261e6eb/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211105192438-b53810dc28af/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2 h1:CIJ76btIcR3eFI5EgSo6k1qKw9KJexJuRLI9G7Hp5wE= golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= +golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 h1:HVyaeDAYux4pnY+D/SiwmLOR36ewZ4iGQIIrtnuCjFA= +golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -1044,12 +1045,15 @@ golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20210806184541-e5e7981a1069/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210816074244-15123e1e1f71/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211106132015-ebca88c72f68/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-20220111092808-5a964db01320/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220114195835-da31bd327af9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220908164124-27713097b956 h1:XeJjHH1KiLpKGb6lvMiksZ9l0fVUh+AmGcm0nOMEBOY= golang.org/x/sys v0.0.0-20220908164124-27713097b956/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/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= @@ -1199,8 +1203,9 @@ google.golang.org/genproto v0.0.0-20200806141610-86f49bd18e98/go.mod h1:FWY/as6D google.golang.org/genproto v0.0.0-20200825200019-8632dd797987/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no= google.golang.org/genproto v0.0.0-20210106152847-07624b53cd92/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no= google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0= -google.golang.org/genproto v0.0.0-20211104193956-4c6863e31247 h1:ZONpjmFT5e+I/0/xE3XXbG5OIvX2hRYzol04MhKBl2E= google.golang.org/genproto v0.0.0-20211104193956-4c6863e31247/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc= +google.golang.org/genproto v0.0.0-20220504150022-98cd25cafc72 h1:iif0mpUetMBqcQPUoq+JnCcmzvfpp8wRx515va8wP1c= +google.golang.org/genproto v0.0.0-20220504150022-98cd25cafc72/go.mod h1:RAyBrSAP7Fh3Nc84ghnVLDPuV51xc9agzmm4Ph6i0Q4= google.golang.org/grpc v1.8.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw= google.golang.org/grpc v1.14.0/go.mod h1:yo6s7OP7yaDglbqo1J04qKzAhqBH6lvTonzMVmEdcZw= google.golang.org/grpc v1.17.0/go.mod h1:6QZJwpn2B+Zp71q/5VxRsJ6NXXVCE5NRUHRo+f3cWCs= @@ -1228,7 +1233,8 @@ google.golang.org/grpc v1.36.0/go.mod h1:qjiiYl8FncCW8feJPdyg3v6XW24KsRHe+dy9BAG google.golang.org/grpc v1.38.0/go.mod h1:NREThFqKR1f3iQ6oBuvc5LadQuXVGo9rkm5ZGrQdJfM= google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34= google.golang.org/grpc v1.41.0/go.mod h1:U3l9uK9J0sini8mHphKoXyaqDA/8VyGnDee1zzIUK6k= -google.golang.org/grpc v1.42.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU= +google.golang.org/grpc v1.46.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk= +google.golang.org/grpc v1.46.2/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk= google.golang.org/grpc v1.48.0/go.mod h1:vN9eftEi1UMyUsIF80+uQXhHjbXYbm0uXoFCACuMGWk= google.golang.org/grpc v1.50.1 h1:DS/BukOZWp8s6p4Dt/tOaJaTQyPyOoCcrjroHuCeLzY= google.golang.org/grpc v1.50.1/go.mod h1:ZgQEeidpAuNRZ8iRrlBKXZQP1ghovWIVhdJRyCDK+GI= @@ -1245,6 +1251,7 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w= google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= diff --git a/registry/polaris/core.go b/registry/polaris/core.go index 7e9f304594..f2b35332a0 100644 --- a/registry/polaris/core.go +++ b/registry/polaris/core.go @@ -23,7 +23,8 @@ import ( ) import ( - "github.com/polarismesh/polaris-go/api" + api "github.com/polarismesh/polaris-go" + internalapi "github.com/polarismesh/polaris-go/api" "github.com/polarismesh/polaris-go/pkg/model" ) @@ -88,7 +89,7 @@ func (watcher *PolarisServiceWatcher) startWatch() { select { case event := <-resp.EventChannel: eType := event.GetSubScribeEventType() - if eType == api.EventInstance { + if eType == internalapi.EventInstance { insEvent := event.(*model.InstanceEvent) if insEvent.AddEvent != nil { diff --git a/registry/polaris/listener.go b/registry/polaris/listener.go index 49ef6dcdd1..c847dd7bc1 100644 --- a/registry/polaris/listener.go +++ b/registry/polaris/listener.go @@ -57,7 +57,7 @@ func NewPolarisListener(watcher *PolarisServiceWatcher) (*polarisListener, error func (pl *polarisListener) startListen() { pl.watcher.AddSubscriber(func(et remoting.EventType, ins []model.Instance) { for i := range ins { - pl.events.In() <- &config_center.ConfigChangeEvent{Value: generateUrl(ins[i]), ConfigType: et} + pl.events.In() <- &config_center.ConfigChangeEvent{Value: ins[i], ConfigType: et} } }) } diff --git a/registry/polaris/registry.go b/registry/polaris/registry.go index 7b0526a77b..02f1567228 100644 --- a/registry/polaris/registry.go +++ b/registry/polaris/registry.go @@ -18,7 +18,6 @@ package polaris import ( - "context" "strconv" "sync" "time" @@ -29,7 +28,7 @@ import ( perrors "github.com/pkg/errors" - "github.com/polarismesh/polaris-go/api" + api "github.com/polarismesh/polaris-go" "github.com/polarismesh/polaris-go/pkg/model" ) @@ -52,98 +51,72 @@ func init() { // newPolarisRegistry will create new instance func newPolarisRegistry(url *common.URL) (registry.Registry, error) { - sdkCtx, ns, err := polaris.GetPolarisConfig(url) - if err != nil { + if err := polaris.InitSDKContext(url); err != nil { return &polarisRegistry{}, err } + + providerApi, err := polaris.GetProviderAPI() + if err != nil { + return nil, err + } + + consumerApi, err := polaris.GetConsumerAPI() + if err != nil { + return nil, err + } + pRegistry := &polarisRegistry{ - namespace: ns, - provider: api.NewProviderAPIByContext(sdkCtx), - consumer: api.NewConsumerAPIByContext(sdkCtx), - lock: &sync.RWMutex{}, - registryUrls: make(map[string]*PolarisHeartbeat), - listenerLock: &sync.RWMutex{}, + url: url, + namespace: url.GetParam(constant.RegistryNamespaceKey, constant.PolarisDefaultNamespace), + provider: providerApi, + consumer: consumerApi, + registryUrls: make([]*common.URL, 0, 4), + watchers: map[string]*PolarisServiceWatcher{}, } return pRegistry, nil } type polarisRegistry struct { - consumer api.ConsumerAPI namespace string url *common.URL + consumer api.ConsumerAPI provider api.ProviderAPI - lock *sync.RWMutex - registryUrls map[string]*PolarisHeartbeat + lock sync.RWMutex + registryUrls []*common.URL + listenerLock sync.RWMutex watchers map[string]*PolarisServiceWatcher - listenerLock *sync.RWMutex } // Register will register the service @url to its polaris registry center. func (pr *polarisRegistry) Register(url *common.URL) error { + serviceName := getServiceName(url) - param := createRegisterParam(url, serviceName) - resp, err := pr.provider.Register(param) + request := createRegisterParam(url, serviceName) + request.Namespace = pr.namespace + resp, err := pr.provider.RegisterInstance(request) if err != nil { return err } if resp.Existed { logger.Warnf("instance already regist, namespace:%+v, service:%+v, host:%+v, port:%+v", - param.Namespace, param.Service, param.Host, param.Port) + request.Namespace, request.Service, request.Host, request.Port) } - - pr.lock.Lock() - defer pr.lock.Unlock() - url.SetParam(constant.PolarisInstanceID, resp.InstanceID) - ctx, cancel := context.WithCancel(context.Background()) - go pr.doHeartbeat(ctx, param) + pr.lock.Lock() + pr.registryUrls = append(pr.registryUrls, url) + pr.lock.Unlock() - pr.registryUrls[url.Key()] = &PolarisHeartbeat{ - url: url, - cancel: cancel, - } return nil } // UnRegister returns nil if unregister successfully. If not, returns an error. func (pr *polarisRegistry) UnRegister(conf *common.URL) error { - var ( - ok bool - err error - oldVal *PolarisHeartbeat - ) - - func() { - pr.lock.Lock() - defer pr.lock.Unlock() - - oldVal, ok = pr.registryUrls[conf.Key()] - - if !ok { - err = perrors.Errorf("Path{%s} has not registered", conf.Key()) - return - } - - oldVal.cancel() - delete(pr.registryUrls, oldVal.url.Key()) - }() - - if err != nil { - return err - } - request := createDeregisterParam(conf, getServiceName(conf)) - - err = pr.provider.Deregister(request) - if err != nil { - func() { - pr.lock.Lock() - defer pr.lock.Unlock() - pr.registryUrls[conf.Key()] = oldVal - }() + request.Namespace = pr.namespace + if err := pr.provider.Deregister(request); err != nil { return perrors.WithMessagef(err, "register(conf:%+v)", conf) } return nil @@ -159,26 +132,17 @@ func (pr *polarisRegistry) Subscribe(url *common.URL, notifyListener registry.No timer := time.NewTimer(time.Duration(RegistryConnDelay) * time.Second) defer timer.Stop() - req := api.WatchServiceRequest{ - WatchServiceRequest: model.WatchServiceRequest{ - Key: model.ServiceKey{ - Service: common.GetSubscribeName(url), - Namespace: pr.namespace, - }, - }, - } - for { - watcher, err := newPolarisWatcher(&req, pr.consumer) - + serviceName := getSubscribeName(url) + watcher, err := pr.createPolarisWatcher(serviceName) if err != nil { logger.Warnf("getwatcher() = err:%v", perrors.WithStack(err)) <-timer.C timer.Reset(time.Duration(RegistryConnDelay) * time.Second) continue } - listener, err := NewPolarisListener(watcher) + listener, err := NewPolarisListener(watcher) if err != nil { logger.Warnf("getListener() = err:%v", perrors.WithStack(err)) <-timer.C @@ -186,13 +150,6 @@ func (pr *polarisRegistry) Subscribe(url *common.URL, notifyListener registry.No continue } - if err != nil { - logger.Warnf("getwatcher() = err:%v", perrors.WithStack(err)) - timer := time.NewTimer(time.Duration(RegistryConnDelay) * time.Second) - timer.Reset(time.Duration(RegistryConnDelay) * time.Second) - continue - } - for { serviceEvent, err := listener.Next() @@ -203,7 +160,6 @@ func (pr *polarisRegistry) Subscribe(url *common.URL, notifyListener registry.No } logger.Infof("update begin, service event: %v", serviceEvent.String()) notifyListener.Notify(serviceEvent) - watcher.startWatch() } } } @@ -246,12 +202,12 @@ func (pr *polarisRegistry) createPolarisWatcher(serviceName string) (*PolarisSer // Destroy stop polaris registry. func (pr *polarisRegistry) Destroy() { - for _, val := range pr.registryUrls { - val.cancel() - err := pr.UnRegister(val.url) - logger.Infof("DeRegister Polaris URL:%+v", val.url) + for i := range pr.registryUrls { + url := pr.registryUrls[i] + err := pr.UnRegister(url) + logger.Infof("DeRegister Polaris URL:%+v", url) if err != nil { - logger.Errorf("Deregister URL:%+v err:%v", val.url, err.Error()) + logger.Errorf("Deregister Polaris URL:%+v err:%v", url, err.Error()) } } return @@ -262,31 +218,6 @@ func (pr *polarisRegistry) IsAvailable() bool { return true } -// doHeartbeat Since polaris does not support automatic reporting of instance heartbeats, separate logic is -// -// needed to implement it -func (pr *polarisRegistry) doHeartbeat(ctx context.Context, ins *api.InstanceRegisterRequest) { - ticker := time.NewTicker(time.Duration(4) * time.Second) - - heartbeat := &api.InstanceHeartbeatRequest{ - InstanceHeartbeatRequest: model.InstanceHeartbeatRequest{ - Service: ins.Service, - Namespace: ins.Namespace, - Host: ins.Host, - Port: ins.Port, - }, - } - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - pr.provider.Heartbeat(heartbeat) - } - } -} - // createRegisterParam convert dubbo url to polaris instance register request func createRegisterParam(url *common.URL, serviceName string) *api.InstanceRegisterRequest { common.HandleRegisterIPAndPort(url) @@ -299,14 +230,16 @@ func createRegisterParam(url *common.URL, serviceName string) *api.InstanceRegis }) metadata[constant.PolarisDubboPath] = url.Path + ver := url.GetParam("version", "") + req := &api.InstanceRegisterRequest{ InstanceRegisterRequest: model.InstanceRegisterRequest{ - Service: serviceName, - Namespace: url.GetParam(constant.PolarisNamespace, constant.PolarisDefaultNamespace), - Host: url.Ip, - Port: port, - Protocol: &protocolForDubboGO, - Metadata: metadata, + Service: serviceName, + Host: url.Ip, + Port: port, + Protocol: &protocolForDubboGO, + Version: &ver, + Metadata: metadata, }, } @@ -321,10 +254,9 @@ func createDeregisterParam(url *common.URL, serviceName string) *api.InstanceDeR port, _ := strconv.Atoi(url.Port) return &api.InstanceDeRegisterRequest{ InstanceDeRegisterRequest: model.InstanceDeRegisterRequest{ - Service: serviceName, - Namespace: url.GetParam(constant.PolarisNamespace, constant.PolarisDefaultNamespace), - Host: url.Ip, - Port: port, + Service: serviceName, + Host: url.Ip, + Port: port, }, } } diff --git a/registry/polaris/service_discovery.go b/registry/polaris/service_discovery.go index c4d6f189be..bb460875ad 100644 --- a/registry/polaris/service_discovery.go +++ b/registry/polaris/service_discovery.go @@ -18,11 +18,8 @@ package polaris import ( - "context" "fmt" - "net/url" "sync" - "time" ) import ( @@ -32,42 +29,60 @@ import ( perrors "github.com/pkg/errors" - "github.com/polarismesh/polaris-go/api" + api "github.com/polarismesh/polaris-go" "github.com/polarismesh/polaris-go/pkg/model" ) import ( "dubbo.apache.org/dubbo-go/v3/common" "dubbo.apache.org/dubbo-go/v3/common/constant" - "dubbo.apache.org/dubbo-go/v3/config" + "dubbo.apache.org/dubbo-go/v3/common/extension" "dubbo.apache.org/dubbo-go/v3/registry" "dubbo.apache.org/dubbo-go/v3/remoting" "dubbo.apache.org/dubbo-go/v3/remoting/polaris" ) +func init() { + extension.SetServiceDiscovery(constant.PolarisKey, newPolarisServiceDiscovery) +} + // newPolarisServiceDiscovery will create new service discovery instance -func newPolarisServiceDiscovery() (registry.ServiceDiscovery, error) { - metadataReportConfig := config.GetMetadataReportConfg() - url := common.NewURLWithOptions( - common.WithParams(make(url.Values)), - common.WithParamsValue(constant.RegistryTimeoutKey, metadataReportConfig.Timeout)) - url.Location = metadataReportConfig.Address +func newPolarisServiceDiscovery(url *common.URL) (registry.ServiceDiscovery, error) { + discoveryURL := common.NewURLWithOptions( + common.WithParams(url.GetParams()), + common.WithParamsValue(constant.TimeoutKey, url.GetParam(constant.RegistryTimeoutKey, constant.DefaultRegTimeout)), + common.WithParamsValue(constant.PolarisServiceToken, url.Password), + common.WithParamsValue(constant.RegistryNamespaceKey, url.GetParam(constant.RegistryNamespaceKey, constant.PolarisDefaultNamespace))) + discoveryURL.Location = url.Location + discoveryURL.Password = url.Password + + if err := polaris.InitSDKContext(url); err != nil { + return nil, err + } + + providerApi, err := polaris.GetProviderAPI() + if err != nil { + return nil, err + } - sdkCtx, namespace, err := polaris.GetPolarisConfig(url) + consumerApi, err := polaris.GetConsumerAPI() + if err != nil { + return nil, err + } if err != nil { return nil, perrors.WithMessage(err, "create polaris namingClient failed.") } - descriptor := fmt.Sprintf("polaris-service-discovery[%s]", metadataReportConfig.Address) + descriptor := fmt.Sprintf("polaris-service-discovery[%s]", discoveryURL.Location) newInstance := &polarisServiceDiscovery{ - namespace: namespace, + namespace: discoveryURL.GetParam(constant.RegistryNamespaceKey, constant.PolarisDefaultNamespace), descriptor: descriptor, instanceLock: &sync.RWMutex{}, - consumer: api.NewConsumerAPIByContext(sdkCtx), - provider: api.NewProviderAPIByContext(sdkCtx), - registryInstances: make(map[string]*PolarisHeartbeat), + consumer: consumerApi, + provider: providerApi, + registryInstances: make(map[string]*PolarisInstanceInfo), listenerLock: &sync.RWMutex{}, watchers: make(map[string]*PolarisServiceWatcher), } @@ -81,7 +96,7 @@ type polarisServiceDiscovery struct { consumer api.ConsumerAPI services *gxset.HashSet instanceLock *sync.RWMutex - registryInstances map[string]*PolarisHeartbeat + registryInstances map[string]*PolarisInstanceInfo watchers map[string]*PolarisServiceWatcher listenerLock *sync.RWMutex } @@ -90,9 +105,6 @@ type polarisServiceDiscovery struct { // and close polaris.ConsumerAPI and polaris.ProviderAPI func (polaris *polarisServiceDiscovery) Destroy() error { for _, inst := range polaris.registryInstances { - - inst.cancel() - err := polaris.Unregister(inst.instance) logger.Infof("Unregister polaris instance:%+v", inst) if err != nil { @@ -108,7 +120,7 @@ func (polaris *polarisServiceDiscovery) Destroy() error { func (polaris *polarisServiceDiscovery) Register(instance registry.ServiceInstance) error { ins := convertToRegisterInstance(polaris.namespace, instance) - resp, err := polaris.provider.Register(ins) + resp, err := polaris.provider.RegisterInstance(ins) if err != nil { return perrors.WithMessage(err, "could not register the instance. "+instance.GetServiceName()) } @@ -118,14 +130,10 @@ func (polaris *polarisServiceDiscovery) Register(instance registry.ServiceInstan polaris.namespace, instance.GetServiceName(), instance.GetHost(), instance.GetPort()) } - ctx, cancel := context.WithCancel(context.Background()) - go polaris.doHeartbeat(ctx, ins) - polaris.instanceLock.Lock() defer polaris.instanceLock.Unlock() - polaris.registryInstances[getInstanceKey(polaris.namespace, instance)] = &PolarisHeartbeat{ - cancel: cancel, + polaris.registryInstances[getInstanceKey(polaris.namespace, instance)] = &PolarisInstanceInfo{ instance: instance, } polaris.services.Add(instance.GetServiceName()) @@ -149,10 +157,8 @@ func (polaris *polarisServiceDiscovery) Unregister(instance registry.ServiceInst func() { polaris.instanceLock.Lock() defer polaris.instanceLock.Unlock() - key := getInstanceKey(polaris.namespace, instance) - if heartbeat, exist := polaris.registryInstances[key]; exist { - heartbeat.cancel() + if _, exist := polaris.registryInstances[key]; exist { delete(polaris.registryInstances, key) } }() @@ -340,27 +346,3 @@ func convertToDeregisterInstance(namespace string, instance registry.ServiceInst }, } } - -// doHeartbeat Since polaris does not support automatic reporting of instance heartbeats, separate logic is -// needed to implement it -func (polaris *polarisServiceDiscovery) doHeartbeat(ctx context.Context, ins *api.InstanceRegisterRequest) { - ticker := time.NewTicker(time.Duration(4) * time.Second) - - heartbeat := &api.InstanceHeartbeatRequest{ - InstanceHeartbeatRequest: model.InstanceHeartbeatRequest{ - Service: ins.Service, - Namespace: ins.Namespace, - Host: ins.Host, - Port: ins.Port, - }, - } - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - polaris.provider.Heartbeat(heartbeat) - } - } -} diff --git a/registry/polaris/utils.go b/registry/polaris/utils.go index 833913145a..39b50a3c58 100644 --- a/registry/polaris/utils.go +++ b/registry/polaris/utils.go @@ -19,7 +19,6 @@ package polaris import ( "bytes" - "context" "fmt" "strconv" "strings" @@ -35,8 +34,7 @@ var ( protocolForDubboGO string = "dubbo" ) -type PolarisHeartbeat struct { - cancel context.CancelFunc +type PolarisInstanceInfo struct { instance registry.ServiceInstance url *common.URL } @@ -61,6 +59,14 @@ func getServiceName(url *common.URL) string { return buffer.String() } +func getSubscribeName(url *common.URL) string { + var buffer bytes.Buffer + + buffer.Write([]byte(common.DubboNodes[common.PROVIDER])) + appendParam(&buffer, url, constant.InterfaceKey) + return buffer.String() +} + // just copy from dubbo-go for nacos func appendParam(target *bytes.Buffer, url *common.URL, key string) { value := url.GetParam(key, "") diff --git a/remoting/polaris/builder.go b/remoting/polaris/builder.go index 0042aa0489..a45fad7d9b 100644 --- a/remoting/polaris/builder.go +++ b/remoting/polaris/builder.go @@ -23,57 +23,93 @@ import ( "net" "strconv" "strings" + "sync" ) import ( perrors "github.com/pkg/errors" + "github.com/polarismesh/polaris-go" "github.com/polarismesh/polaris-go/api" "github.com/polarismesh/polaris-go/pkg/config" - "github.com/polarismesh/polaris-go/pkg/model" ) import ( "dubbo.apache.org/dubbo-go/v3/common" - "dubbo.apache.org/dubbo-go/v3/common/constant" ) -// GetPolarisConfig get polaris config from dubbo url -func GetPolarisConfig(url *common.URL) (api.SDKContext, string, error) { - if url == nil { - return nil, "", errors.New("url is empty!") +var ( + once sync.Once + sdkCtx api.SDKContext +) + +var ( + ErrorSDKContextNotInit = errors.New("polaris SDKContext not init") +) + +// GetConsumerAPI creates one polaris ConsumerAPI instance +func GetConsumerAPI() (polaris.ConsumerAPI, error) { + if sdkCtx == nil { + return nil, ErrorSDKContextNotInit } - addresses := strings.Split(url.Location, ",") - serverConfigs := make([]string, 0, len(addresses)) - for _, addr := range addresses { - ip, portStr, err := net.SplitHostPort(addr) - if err != nil { - return nil, "", perrors.WithMessagef(err, "split [%s] ", addr) - } - port, _ := strconv.Atoi(portStr) - serverConfigs = append(serverConfigs, fmt.Sprintf("%s:%d", ip, uint64(port))) + return polaris.NewConsumerAPIByContext(sdkCtx), nil +} + +// GetProviderAPI creates one polaris ProviderAPI instance +func GetProviderAPI() (polaris.ProviderAPI, error) { + if sdkCtx == nil { + return nil, ErrorSDKContextNotInit } - polarisConf := config.NewDefaultConfiguration(serverConfigs) + return polaris.NewProviderAPIByContext(sdkCtx), nil +} - confPath := url.GetParam(constant.PolarisConfigFilePath, "") - if confPath != "" && model.IsFile(confPath) { - complexConf, err := config.LoadConfigurationByFile(confPath) - if err != nil { - return nil, "", err - } - mergePolarisConfiguration(polarisConf, complexConf) +// GetRouterAPI create one polaris RouterAPI instance +func GetRouterAPI() (polaris.RouterAPI, error) { + if sdkCtx == nil { + return nil, ErrorSDKContextNotInit + } + + return polaris.NewRouterAPIByContext(sdkCtx), nil +} - polarisConf = complexConf +// GetLimiterAPI creates one polaris LimiterAPI instance +func GetLimiterAPI() (polaris.LimitAPI, error) { + if sdkCtx == nil { + return nil, ErrorSDKContextNotInit } - sdkCtx, err := api.InitContextByConfig(polarisConf) - if err != nil { - return nil, "", err + return polaris.NewLimitAPIByContext(sdkCtx), nil +} + +// InitSDKContext inits polaris SDKContext by URL +func InitSDKContext(url *common.URL) error { + if url == nil { + return errors.New("url is empty!") } - return sdkCtx, url.GetParam(constant.PolarisNamespace, constant.PolarisDefaultNamespace), nil + var rerr error + once.Do(func() { + + addresses := strings.Split(url.Location, ",") + serverConfigs := make([]string, 0, len(addresses)) + for _, addr := range addresses { + ip, portStr, err := net.SplitHostPort(addr) + if err != nil { + rerr = perrors.WithMessagef(err, "split [%s] ", addr) + } + port, _ := strconv.Atoi(portStr) + serverConfigs = append(serverConfigs, fmt.Sprintf("%s:%d", ip, uint64(port))) + } + + polarisConf := config.NewDefaultConfiguration(serverConfigs) + _sdkCtx, err := api.InitContextByConfig(polarisConf) + rerr = err + sdkCtx = _sdkCtx + }) + + return rerr } func mergePolarisConfiguration(easy, complexConf config.Configuration) { diff --git a/remoting/polaris/builder_test.go b/remoting/polaris/builder_test.go index 2cf4898361..6a27f33074 100644 --- a/remoting/polaris/builder_test.go +++ b/remoting/polaris/builder_test.go @@ -28,79 +28,18 @@ import ( import ( "dubbo.apache.org/dubbo-go/v3/common" - "dubbo.apache.org/dubbo-go/v3/common/constant" - "dubbo.apache.org/dubbo-go/v3/config" ) -func TestGetPolarisConfig(t *testing.T) { - - rc := &config.RemoteConfig{} - rc.Params = make(map[string]string) - - rc.Protocol = "polaris" - rc.Address = "127.0.0.1:8091" - - rc.Params[constant.PolarisNamespace] = "default" - - url, err := rc.ToURL() - if err != nil { - t.Fatal(err) - } - - sdkCtx, namespace, err := GetPolarisConfig(url) - - if err != nil { - t.Fatal(err) - } - assert.NotNil(t, sdkCtx, "SDKContext") - - assert.Equal(t, "default", namespace, "namespace") - assert.ElementsMatch(t, []string{"127.0.0.1:8091"}, sdkCtx.GetConfig().GetGlobal().GetServerConnector().GetAddresses(), "server address") -} - -func TestGetPolarisConfigWithExternalFile(t *testing.T) { - - rc := &config.RemoteConfig{} - rc.Params = make(map[string]string) - - rc.Protocol = "polaris" - rc.Address = "127.0.0.1:8091" - - rc.Params[constant.PolarisNamespace] = "default" - rc.Params[constant.PolarisConfigFilePath] = "./polaris.yaml" - - url, err := rc.ToURL() - if err != nil { - t.Fatal(err) - } - - sdkCtx, namespace, err := GetPolarisConfig(url) - - if err != nil { - t.Fatal(err) - } - assert.NotNil(t, sdkCtx, "SDKContext") - - assert.Equal(t, "default", namespace, "namespace") - assert.ElementsMatch(t, []string{"127.0.0.1:8091", "127.0.0.2:8091"}, sdkCtx.GetConfig().GetGlobal().GetServerConnector().GetAddresses(), "server address") -} - func TestGetPolarisConfigByUrl(t *testing.T) { regurl := getRegUrl() - sdkCtx, namespace, err := GetPolarisConfig(regurl) + err := InitSDKContext(regurl) assert.Nil(t, err) - assert.Equal(t, "default", namespace, "namespace") - assert.ElementsMatch(t, []string{"127.0.0.1:8091", "127.0.0.2:8091"}, sdkCtx.GetConfig().GetGlobal().GetServerConnector().GetAddresses(), "server address") + assert.ElementsMatch(t, []string{"127.0.0.1:8091"}, sdkCtx.GetConfig().GetGlobal().GetServerConnector().GetAddresses(), "server address") } func getRegUrl() *common.URL { - regurlMap := url.Values{} - regurlMap.Set(constant.PolarisNamespace, "default") - regurlMap.Set(constant.PolarisConfigFilePath, "./polaris.yaml") - regurl, _ := common.NewURL("registry://127.0.0.1:8091", common.WithParams(regurlMap)) - return regurl } diff --git a/remoting/polaris/polaris.yaml b/remoting/polaris/polaris.yaml index 3d794ec552..2347aad371 100644 --- a/remoting/polaris/polaris.yaml +++ b/remoting/polaris/polaris.yaml @@ -19,8 +19,6 @@ global: maxRetryTimes: 5 retryInterval: 1s serverConnector: - addresses: - - 127.0.0.2:8091 protocol: grpc connectTimeout: 500ms messageTimeout: 1s