This is an automated email from the ASF dual-hosted git repository. lujiajing pushed a commit to branch node-selector in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit f3c6e2e442246927f16f47a06a845f34d3260911 Author: Megrez Lu <[email protected]> AuthorDate: Wed Sep 20 18:00:19 2023 +0800 add node selector --- banyand/liaison/grpc/node.go | 60 +++--- dist/LICENSE | 7 + .../licenses/license-github.com-dchest-siphash.txt | 121 +++++++++++++ dist/licenses/license-github.com-kkdai-maglev.txt | 201 +++++++++++++++++++++ go.mod | 3 + go.sum | 5 + pkg/cmdsetup/liaison.go | 7 +- pkg/node/interface.go | 98 ++++++++++ pkg/node/maglev.go | 53 ++++++ pkg/node/maglev_test.go | 48 +++++ 10 files changed, 573 insertions(+), 30 deletions(-) diff --git a/banyand/liaison/grpc/node.go b/banyand/liaison/grpc/node.go index 73529604..73d69b95 100644 --- a/banyand/liaison/grpc/node.go +++ b/banyand/liaison/grpc/node.go @@ -18,7 +18,6 @@ package grpc import ( - "sort" "sync" "github.com/pkg/errors" @@ -26,6 +25,7 @@ import ( databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" "github.com/apache/skywalking-banyandb/banyand/metadata/schema" "github.com/apache/skywalking-banyandb/banyand/queue" + "github.com/apache/skywalking-banyandb/pkg/node" ) var ( @@ -37,46 +37,47 @@ var ( // together with the shardID calculated from the incoming data. type NodeRegistry interface { Locate(group, name string, shardID uint32) (string, error) + Initialize() error } type clusterNodeService struct { - metaRepo queue.Client - nodes []string - nodeMutex sync.RWMutex + pipeline queue.Client + sel node.Selector sync.Once } // NewClusterNodeRegistry creates a cluster node registry. -func NewClusterNodeRegistry(metaRepo queue.Client) NodeRegistry { +func NewClusterNodeRegistry(pipeline queue.Client, selector node.Selector) NodeRegistry { nr := &clusterNodeService{ - metaRepo: metaRepo, + pipeline: pipeline, + sel: selector, } - metaRepo.Register(nr) return nr } -func (n *clusterNodeService) Locate(_, _ string, shardID uint32) (string, error) { - // Use round-robin to select the node. - n.nodeMutex.RLock() - defer n.nodeMutex.RUnlock() - if len(n.nodes) == 0 { - return "", errors.New("no node available") +func (n *clusterNodeService) Initialize() error { + n.Do(func() { + n.pipeline.Register(n) + }) + return nil +} + +func (n *clusterNodeService) Locate(group, name string, shardID uint32) (string, error) { + nodeID, err := n.sel.Pick(group, name, shardID) + if err != nil { + return "", errors.Wrapf(err, "fail to locate %s/%s(%d)", group, name, shardID) } - return n.nodes[shardID%uint32(len(n.nodes))], nil + return nodeID, nil } func (n *clusterNodeService) OnAddOrUpdate(metadata schema.Metadata) { switch metadata.Kind { case schema.KindNode: - n.nodeMutex.Lock() - defer n.nodeMutex.Unlock() - for _, node := range n.nodes { - if node == metadata.Spec.(*databasev1.Node).Metadata.Name { - return - } + inputNode := metadata.Spec.(*databasev1.Node) + if inputNode.Metadata.GetName() == "" { + return } - n.nodes = append(n.nodes, metadata.Spec.(*databasev1.Node).Metadata.Name) - sort.Strings(n.nodes) + n.sel.AddNode(inputNode) default: } } @@ -84,14 +85,11 @@ func (n *clusterNodeService) OnAddOrUpdate(metadata schema.Metadata) { func (n *clusterNodeService) OnDelete(metadata schema.Metadata) { switch metadata.Kind { case schema.KindNode: - n.nodeMutex.Lock() - defer n.nodeMutex.Unlock() - for i, node := range n.nodes { - if node == metadata.Spec.(*databasev1.Node).Metadata.Name { - n.nodes = append(n.nodes[:i], n.nodes[i+1:]...) - break - } + dNode := metadata.Spec.(*databasev1.Node) + if dNode.Metadata.GetName() == "" { + return } + n.sel.RemoveNode(dNode) default: } } @@ -103,6 +101,10 @@ func NewLocalNodeRegistry() NodeRegistry { return localNodeService{} } +func (localNodeService) Initialize() error { + return nil +} + // Locate of localNodeService always returns local. func (localNodeService) Locate(_, _ string, _ uint32) (string, error) { return "local", nil diff --git a/dist/LICENSE b/dist/LICENSE index 87b47d39..f479b15f 100644 --- a/dist/LICENSE +++ b/dist/LICENSE @@ -198,6 +198,7 @@ Apache-2.0 licenses github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 Apache-2.0 github.com/inconshreveable/mousetrap v1.1.0 Apache-2.0 github.com/jonboulle/clockwork v0.4.0 Apache-2.0 + github.com/kkdai/maglev v0.2.0 Apache-2.0 github.com/matttproud/golang_protobuf_extensions v1.0.4 Apache-2.0 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd Apache-2.0 github.com/modern-go/reflect2 v1.0.2 Apache-2.0 @@ -296,6 +297,12 @@ BSD-3-Clause and Apache-2.0 and MIT licenses github.com/klauspost/compress v1.15.6 BSD-3-Clause and Apache-2.0 and MIT +======================================================================== +CC0-1.0 licenses +======================================================================== + + github.com/dchest/siphash v1.2.3 CC0-1.0 + ======================================================================== ISC licenses ======================================================================== diff --git a/dist/licenses/license-github.com-dchest-siphash.txt b/dist/licenses/license-github.com-dchest-siphash.txt new file mode 100644 index 00000000..0e259d42 --- /dev/null +++ b/dist/licenses/license-github.com-dchest-siphash.txt @@ -0,0 +1,121 @@ +Creative Commons Legal Code + +CC0 1.0 Universal + + CREATIVE COMMONS CORPORATION IS NOT A LAW FIRM AND DOES NOT PROVIDE + LEGAL SERVICES. DISTRIBUTION OF THIS DOCUMENT DOES NOT CREATE AN + ATTORNEY-CLIENT RELATIONSHIP. CREATIVE COMMONS PROVIDES THIS + INFORMATION ON AN "AS-IS" BASIS. CREATIVE COMMONS MAKES NO WARRANTIES + REGARDING THE USE OF THIS DOCUMENT OR THE INFORMATION OR WORKS + PROVIDED HEREUNDER, AND DISCLAIMS LIABILITY FOR DAMAGES RESULTING FROM + THE USE OF THIS DOCUMENT OR THE INFORMATION OR WORKS PROVIDED + HEREUNDER. + +Statement of Purpose + +The laws of most jurisdictions throughout the world automatically confer +exclusive Copyright and Related Rights (defined below) upon the creator +and subsequent owner(s) (each and all, an "owner") of an original work of +authorship and/or a database (each, a "Work"). + +Certain owners wish to permanently relinquish those rights to a Work for +the purpose of contributing to a commons of creative, cultural and +scientific works ("Commons") that the public can reliably and without fear +of later claims of infringement build upon, modify, incorporate in other +works, reuse and redistribute as freely as possible in any form whatsoever +and for any purposes, including without limitation commercial purposes. +These owners may contribute to the Commons to promote the ideal of a free +culture and the further production of creative, cultural and scientific +works, or to gain reputation or greater distribution for their Work in +part through the use and efforts of others. + +For these and/or other purposes and motivations, and without any +expectation of additional consideration or compensation, the person +associating CC0 with a Work (the "Affirmer"), to the extent that he or she +is an owner of Copyright and Related Rights in the Work, voluntarily +elects to apply CC0 to the Work and publicly distribute the Work under its +terms, with knowledge of his or her Copyright and Related Rights in the +Work and the meaning and intended legal effect of CC0 on those rights. + +1. Copyright and Related Rights. A Work made available under CC0 may be +protected by copyright and related or neighboring rights ("Copyright and +Related Rights"). Copyright and Related Rights include, but are not +limited to, the following: + + i. the right to reproduce, adapt, distribute, perform, display, + communicate, and translate a Work; + ii. moral rights retained by the original author(s) and/or performer(s); +iii. publicity and privacy rights pertaining to a person's image or + likeness depicted in a Work; + iv. rights protecting against unfair competition in regards to a Work, + subject to the limitations in paragraph 4(a), below; + v. rights protecting the extraction, dissemination, use and reuse of data + in a Work; + vi. database rights (such as those arising under Directive 96/9/EC of the + European Parliament and of the Council of 11 March 1996 on the legal + protection of databases, and under any national implementation + thereof, including any amended or successor version of such + directive); and +vii. other similar, equivalent or corresponding rights throughout the + world based on applicable law or treaty, and any national + implementations thereof. + +2. Waiver. To the greatest extent permitted by, but not in contravention +of, applicable law, Affirmer hereby overtly, fully, permanently, +irrevocably and unconditionally waives, abandons, and surrenders all of +Affirmer's Copyright and Related Rights and associated claims and causes +of action, whether now known or unknown (including existing as well as +future claims and causes of action), in the Work (i) in all territories +worldwide, (ii) for the maximum duration provided by applicable law or +treaty (including future time extensions), (iii) in any current or future +medium and for any number of copies, and (iv) for any purpose whatsoever, +including without limitation commercial, advertising or promotional +purposes (the "Waiver"). Affirmer makes the Waiver for the benefit of each +member of the public at large and to the detriment of Affirmer's heirs and +successors, fully intending that such Waiver shall not be subject to +revocation, rescission, cancellation, termination, or any other legal or +equitable action to disrupt the quiet enjoyment of the Work by the public +as contemplated by Affirmer's express Statement of Purpose. + +3. Public License Fallback. Should any part of the Waiver for any reason +be judged legally invalid or ineffective under applicable law, then the +Waiver shall be preserved to the maximum extent permitted taking into +account Affirmer's express Statement of Purpose. In addition, to the +extent the Waiver is so judged Affirmer hereby grants to each affected +person a royalty-free, non transferable, non sublicensable, non exclusive, +irrevocable and unconditional license to exercise Affirmer's Copyright and +Related Rights in the Work (i) in all territories worldwide, (ii) for the +maximum duration provided by applicable law or treaty (including future +time extensions), (iii) in any current or future medium and for any number +of copies, and (iv) for any purpose whatsoever, including without +limitation commercial, advertising or promotional purposes (the +"License"). The License shall be deemed effective as of the date CC0 was +applied by Affirmer to the Work. Should any part of the License for any +reason be judged legally invalid or ineffective under applicable law, such +partial invalidity or ineffectiveness shall not invalidate the remainder +of the License, and in such case Affirmer hereby affirms that he or she +will not (i) exercise any of his or her remaining Copyright and Related +Rights in the Work or (ii) assert any associated claims and causes of +action with respect to the Work, in either case contrary to Affirmer's +express Statement of Purpose. + +4. Limitations and Disclaimers. + + a. No trademark or patent rights held by Affirmer are waived, abandoned, + surrendered, licensed or otherwise affected by this document. + b. Affirmer offers the Work as-is and makes no representations or + warranties of any kind concerning the Work, express, implied, + statutory or otherwise, including without limitation warranties of + title, merchantability, fitness for a particular purpose, non + infringement, or the absence of latent or other defects, accuracy, or + the present or absence of errors, whether or not discoverable, all to + the greatest extent permissible under applicable law. + c. Affirmer disclaims responsibility for clearing rights of other persons + that may apply to the Work or any use thereof, including without + limitation any person's Copyright and Related Rights in the Work. + Further, Affirmer disclaims responsibility for obtaining any necessary + consents, permissions or other rights required for any use of the + Work. + d. Affirmer understands and acknowledges that Creative Commons is not a + party to this document and has no duty or obligation with respect to + this CC0 or use of the Work. diff --git a/dist/licenses/license-github.com-kkdai-maglev.txt b/dist/licenses/license-github.com-kkdai-maglev.txt new file mode 100644 index 00000000..8dada3ed --- /dev/null +++ b/dist/licenses/license-github.com-kkdai-maglev.txt @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "{}" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright {yyyy} {name of copyright owner} + + 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. diff --git a/go.mod b/go.mod index 2a013209..991bfa80 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 github.com/hashicorp/golang-lru v0.5.4 github.com/hashicorp/golang-lru/v2 v2.0.4 + github.com/kkdai/maglev v0.2.0 github.com/oklog/run v1.1.0 github.com/onsi/ginkgo/v2 v2.11.0 github.com/onsi/gomega v1.27.10 @@ -44,6 +45,8 @@ require ( sigs.k8s.io/yaml v1.3.0 ) +require github.com/dchest/siphash v1.2.3 // indirect + require ( github.com/axiomhq/hyperloglog v0.0.0-20191112132149-a4c4c47bc57f // indirect github.com/beorn7/perks v1.0.1 // indirect diff --git a/go.sum b/go.sum index eb7ac7e6..29be62ee 100644 --- a/go.sum +++ b/go.sum @@ -104,6 +104,9 @@ github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46t 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= +github.com/dchest/siphash v1.2.2/go.mod h1:q+IRvb2gOSrUnYoPqHiyHXS0FOBBOdl6tONBlVnOnt4= +github.com/dchest/siphash v1.2.3 h1:QXwFc8cFOR2dSa/gE6o/HokBMWtLUaNDVd+22aKHeEA= +github.com/dchest/siphash v1.2.3/go.mod h1:0NvQU092bT0ipiFN++/rXm69QG9tVxLAlQHIXMPAkHc= github.com/dgraph-io/ristretto v0.1.1 h1:6CWw5tJNgpegArSHpNHJKldNeq03FQCwYvfMVWajOK8= github.com/dgraph-io/ristretto v0.1.1/go.mod h1:S1GPSBCYCIhmVNfcth17y2zZtQT6wzkzgwUve0VDWWA= github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA= @@ -269,6 +272,8 @@ github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/X github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/kkdai/maglev v0.2.0 h1:w6DCW0kAA6fstZqXkrBrlgIC3jeIRXkjOYea/m6EK/Y= +github.com/kkdai/maglev v0.2.0/go.mod h1:d+mt8Lmt3uqi9aRb/BnPjzD0fy+ETs1vVXiGRnqHVZ4= github.com/klauspost/compress v1.12.3/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= github.com/klauspost/compress v1.15.2/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= github.com/klauspost/compress v1.15.6 h1:6D9PcO8QWu0JyaQ2zUMmu16T1T+zjjEpP91guRsvDfY= diff --git a/pkg/cmdsetup/liaison.go b/pkg/cmdsetup/liaison.go index 126f4b89..b0c03985 100644 --- a/pkg/cmdsetup/liaison.go +++ b/pkg/cmdsetup/liaison.go @@ -32,6 +32,7 @@ import ( "github.com/apache/skywalking-banyandb/banyand/queue" "github.com/apache/skywalking-banyandb/banyand/queue/pub" "github.com/apache/skywalking-banyandb/pkg/logger" + "github.com/apache/skywalking-banyandb/pkg/node" "github.com/apache/skywalking-banyandb/pkg/run" "github.com/apache/skywalking-banyandb/pkg/version" ) @@ -45,7 +46,11 @@ func newLiaisonCmd(runners ...run.Unit) *cobra.Command { } pipeline := pub.New(metaSvc) localPipeline := queue.Local() - grpcServer := grpc.NewServer(ctx, pipeline, localPipeline, metaSvc, grpc.NewClusterNodeRegistry(pipeline)) + nodeSel, err := node.NewMaglevSelector() + if err != nil { + l.Fatal().Err(err).Msg("failed to initiate required node selector") + } + grpcServer := grpc.NewServer(ctx, pipeline, localPipeline, metaSvc, grpc.NewClusterNodeRegistry(pipeline, nodeSel)) profSvc := observability.NewProfService() metricSvc := observability.NewMetricService() httpServer := http.NewServer() diff --git a/pkg/node/interface.go b/pkg/node/interface.go new file mode 100644 index 00000000..baf91a74 --- /dev/null +++ b/pkg/node/interface.go @@ -0,0 +1,98 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you 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 node provides node selector for liaison. +package node + +import ( + "strconv" + "sync" + + "github.com/pkg/errors" + "golang.org/x/exp/slices" + + databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" +) + +var ( + _ Selector = (*pickFirstSelector)(nil) + + // ErrNoAvailableNode will be returned if no node is available. + ErrNoAvailableNode = errors.New("selector: no available node") +) + +// Selector keeps all data nodes in the memory and can provide different algorithm to pick an available node. +type Selector interface { + AddNode(node *databasev1.Node) + RemoveNode(node *databasev1.Node) + Pick(group, name string, shardID uint32) (string, error) +} + +// NewPickFirstSelector returns a simple selector that always returns the first node if exists. +func NewPickFirstSelector() (Selector, error) { + return &pickFirstSelector{}, nil +} + +// pickFirstSelector always pick the first node in the sorted node ids list. +type pickFirstSelector struct { + nodeIDMap map[string]struct{} + nodeIds []string + mu sync.RWMutex +} + +func (p *pickFirstSelector) AddNode(node *databasev1.Node) { + nodeID := node.GetMetadata().GetName() + p.mu.RLock() + if _, ok := p.nodeIDMap[nodeID]; !ok { + p.mu.RUnlock() + return + } + p.mu.RUnlock() + p.mu.Lock() + defer p.mu.Unlock() + p.nodeIDMap[nodeID] = struct{}{} + p.nodeIds = append(p.nodeIds, nodeID) + slices.Sort(p.nodeIds) +} + +func (p *pickFirstSelector) RemoveNode(node *databasev1.Node) { + nodeID := node.GetMetadata().GetName() + p.mu.RLock() + if _, ok := p.nodeIDMap[nodeID]; !ok { + p.mu.RUnlock() + return + } + p.mu.RUnlock() + p.mu.Lock() + defer p.mu.Unlock() + p.nodeIDMap[nodeID] = struct{}{} + p.nodeIds = append(p.nodeIds, nodeID) + slices.Sort(p.nodeIds) +} + +func (p *pickFirstSelector) Pick(_, _ string, _ uint32) (string, error) { + p.mu.RLock() + defer p.mu.RUnlock() + if len(p.nodeIds) == 0 { + return "", ErrNoAvailableNode + } + return p.nodeIds[0], nil +} + +func formatSearchKey(group, name string, shardID uint32) string { + return group + "/" + name + "#" + strconv.FormatUint(uint64(shardID), 10) +} diff --git a/pkg/node/maglev.go b/pkg/node/maglev.go new file mode 100644 index 00000000..deb140f2 --- /dev/null +++ b/pkg/node/maglev.go @@ -0,0 +1,53 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you 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 node + +import ( + "github.com/kkdai/maglev" + + databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" +) + +var _ Selector = (*maglevSelector)(nil) + +type maglevSelector struct { + maglev *maglev.Maglev +} + +func (m *maglevSelector) AddNode(node *databasev1.Node) { + _ = m.maglev.Add(node.GetMetadata().GetName()) +} + +func (m *maglevSelector) RemoveNode(node *databasev1.Node) { + _ = m.maglev.Remove(node.GetMetadata().GetName()) +} + +func (m *maglevSelector) Pick(group, name string, shardID uint32) (string, error) { + return m.maglev.Get(formatSearchKey(group, name, shardID)) +} + +// NewMaglevSelector creates a new backend selector based on Maglev hashing algorithm. +func NewMaglevSelector() (Selector, error) { + alg, err := maglev.NewMaglev(nil, 65537) + if err != nil { + return nil, err + } + return &maglevSelector{ + maglev: alg, + }, nil +} diff --git a/pkg/node/maglev_test.go b/pkg/node/maglev_test.go new file mode 100644 index 00000000..b0bb0921 --- /dev/null +++ b/pkg/node/maglev_test.go @@ -0,0 +1,48 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you 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 node + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1" + databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1" +) + +func TestMaglevSelector(t *testing.T) { + sel, err := NewMaglevSelector() + assert.NoError(t, err) + sel.AddNode(&databasev1.Node{ + Metadata: &commonv1.Metadata{ + Name: "data-node-1", + }, + }) + sel.AddNode(&databasev1.Node{ + Metadata: &commonv1.Metadata{ + Name: "data-node-2", + }, + }) + nodeID1, err := sel.Pick("sw_metrics", "traffic_instance", 0) + assert.NoError(t, err) + assert.Contains(t, []string{"data-node-1", "data-node-2"}, nodeID1) + nodeID2, err := sel.Pick("sw_metrics", "traffic_instance", 0) + assert.NoError(t, err) + assert.Equal(t, nodeID2, nodeID1) +}
