Skip to content

Commit e37cf23

Browse files
derekwinderekwin
committed
add locality loadbalance
1 parent 20cb2d8 commit e37cf23

File tree

7 files changed

+284
-27
lines changed

7 files changed

+284
-27
lines changed

‎bpf/kmesh/workload/include/service.h‎

Lines changed: 51 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,13 @@ static inline int lb_random_handle(struct kmesh_context *kmesh_ctx, __u32 servic
1717
int ret = 0;
1818
endpoint_key endpoint_k = {0};
1919
endpoint_value *endpoint_v = NULL;
20+
int rand_k = 0;
2021

2122
endpoint_k.service_id = service_id;
22-
endpoint_k.backend_index = bpf_get_prandom_u32() % service_v->endpoint_count + 1;
23+
endpoint_k.prio = MAX_PRIO; // for random handle，all endpoints are saved in MAX_PRIO
24+
25+
rand_k = bpf_get_prandom_u32() % service_v->prio_endpoint_count[MAX_PRIO]+1;
26+
endpoint_k.backend_index = rand_k;
2327

2428
endpoint_v = map_lookup_endpoint(&endpoint_k);
2529
if (!endpoint_v) {
@@ -37,6 +41,45 @@ static inline int lb_random_handle(struct kmesh_context *kmesh_ctx, __u32 servic
3741
return 0;
3842
}
3943

44+
static inline int lb_locality_failover_handle(struct kmesh_context *kmesh_ctx, __u32 service_id, service_value *service_v, bool is_strict)
45+
{
46+
int ret = 0;
47+
uint32_t rand_k = 0;
48+
endpoint_key endpoint_k = {0};
49+
endpoint_value *endpoint_v = NULL;
50+
endpoint_k.service_id = service_id;
51+
52+
// #pragma unroll
53+
for (int match_rank = MAX_PRIO; match_rank>=0; match_rank--) {
54+
endpoint_k.prio = match_rank; // 6->0
55+
// if we have endpoints in this prio
56+
if (service_v->prio_endpoint_count[match_rank] > 0) {
57+
rand_k = bpf_get_prandom_u32() % service_v->prio_endpoint_count[match_rank]+1;
58+
if (rand_k >= MAP_SIZE_OF_BACKEND) {
59+
return -ENOENT;
60+
}
61+
endpoint_k.backend_index = rand_k;
62+
endpoint_v = map_lookup_endpoint(&endpoint_k);
63+
if (!endpoint_v) {
64+
BPF_LOG(ERR, SERVICE, "find endpoint [%u/%u/%u] failed", service_id, match_rank, endpoint_k.backend_index);
65+
return -ENOENT;
66+
}
67+
ret = endpoint_manager(kmesh_ctx, endpoint_v, service_id, service_v);
68+
if (ret != 0) {
69+
if (ret != -ENOENT)
70+
BPF_LOG(ERR, SERVICE, "endpoint_manager failed, ret:%d\n", ret);
71+
return ret;
72+
}
73+
return 0; // find the backend successfully
74+
}
75+
if (is_strict && match_rank == service_v->lb_strict_index) { // only match lb strict index
76+
return -ENOENT;
77+
}
78+
}
79+
// no backend matched
80+
return -ENOENT;
81+
}
82+
4083
static inline int service_manager(struct kmesh_context *kmesh_ctx, __u32 service_id, service_value *service_v)
4184
{
4285
int ret = 0;
@@ -55,15 +98,18 @@ static inline int service_manager(struct kmesh_context *kmesh_ctx, __u32 service
5598
return ret;
5699
}
57100

58-
if (service_v->endpoint_count == 0) {
59-
BPF_LOG(DEBUG, SERVICE, "service %u has no endpoint", service_id);
60-
return 0;
61-
}
101+
BPF_LOG(DEBUG, SERVICE, "service [%u] policy [%u] failed", service_id, service_v->lb_policy);
62102

63103
switch (service_v->lb_policy) {
64104
case LB_POLICY_RANDOM:
65105
ret = lb_random_handle(kmesh_ctx, service_id, service_v);
66106
break;
107+
case LB_POLICY_STRICT:
108+
ret = lb_locality_failover_handle(kmesh_ctx, service_id, service_v, true);
109+
break;
110+
case LB_POLICY_FAILOVER:
111+
ret = lb_locality_failover_handle(kmesh_ctx, service_id, service_v, false);
112+
break;
67113
default:
68114
BPF_LOG(ERR, SERVICE, "unsupported load balance type:%u\n", service_v->lb_policy);
69115
ret = -EINVAL;

‎bpf/kmesh/workload/include/workload.h‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@
1010
#define MAX_PORT_COUNT 10
1111
#define MAX_SERVICE_COUNT 10
1212
#define RINGBUF_SIZE (1 << 12)
13+
#define MAX_PRIO 6
14+
#define MAX_PRIO_COUNT MAX_PRIO+1 // 6 means match all scope, 0 means match nothing
1315
#define MAX_MEMBER_NUM_PER_POLICY 4
1416

1517
#pragma pack(1)
@@ -28,8 +30,9 @@ typedef struct {
2830
} service_key;
2931

3032
typedef struct {
31-
__u32 endpoint_count; // endpoint count of current service
32-
__u32 lb_policy; // load balancing algorithm, currently only supports random algorithm
33+
__u32 prio_endpoint_count[MAX_PRIO_COUNT];// endpoint count of current service with prio, prio from 6->0
34+
__u32 lb_policy; // load balancing algorithm, currently supports random algorithm, locality loadbalance Failover/strict mode
35+
__u32 lb_strict_index; // for failover strict mode
3336
__u32 service_port[MAX_PORT_COUNT]; // service_port[i] and target_port[i] are a pair, i starts from 0 and max value
3437
// is MAX_PORT_COUNT-1
3538
__u32 target_port[MAX_PORT_COUNT];
@@ -40,6 +43,7 @@ typedef struct {
4043
// endpoint map
4144
typedef struct {
4245
__u32 service_id; // service id
46+
__u32 prio; // prio means rank, 6 means match all, and 0 means match nothing
4347
__u32 backend_index; // if endpoint_count = 3, then backend_index = 0/1/2
4448
} endpoint_key;
4549

‎bpf/kmesh/workload/include/workload_common.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
// loadbalance type
2323
typedef enum {
2424
LB_POLICY_RANDOM = 0,
25+
LB_POLICY_STRICT = 1,
26+
LB_POLICY_FAILOVER = 2,
2527
} lb_policy_t;
2628

2729
#pragma pack(1)

‎pkg/controller/workload/bpfcache/endpoint.go‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,14 @@ import (
2121
"istio.io/istio/pkg/util/sets"
2222
)
2323

24+
const (
25+
MaxPrio = 6
26+
MaxPrioNum = 7
27+
)
28+
2429
type EndpointKey struct {
2530
ServiceId uint32 // service id
31+
Prio uint32
2632
BackendIndex uint32 // if endpoint_count = 3, then backend_index = 1/2/3
2733
}
2834

@@ -59,15 +65,17 @@ func (c *Cache) EndpointDelete(key *EndpointKey) error {
5965
}
6066

6167
// EndpointSwap update the last endpoint index and remove the current endpoint
62-
func (c *Cache) EndpointSwap(currentIndex, lastIndex uint32, serviceId uint32) error {
68+
func (c *Cache) EndpointSwap(currentIndex, lastIndex uint32, serviceId uint32, prio uint32) error {
6369
if currentIndex == lastIndex {
6470
return c.EndpointDelete(&EndpointKey{
6571
ServiceId: serviceId,
72+
Prio: prio,
6673
BackendIndex: lastIndex,
6774
})
6875
}
6976
lastKey := &EndpointKey{
7077
ServiceId: serviceId,
78+
Prio: prio,
7179
BackendIndex: lastIndex,
7280
}
7381
lastValue := &EndpointValue{}
@@ -77,6 +85,7 @@ func (c *Cache) EndpointSwap(currentIndex, lastIndex uint32, serviceId uint32) e
7785

7886
currentKey := &EndpointKey{
7987
ServiceId: serviceId,
88+
Prio: prio,
8089
BackendIndex: currentIndex,
8190
}
8291
currentValue := &EndpointValue{}

‎pkg/controller/workload/bpfcache/locality_cache.go‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
package bpfcache
2+
3+
import (
4+
"kmesh.net/kmesh/api/v2/workloadapi"
5+
)
6+
7+
const (
8+
REGION = 1 << iota // 000001
9+
ZONE // 000010
10+
SUBZONE // 000100
11+
NODENAME // 001000
12+
CLUSTERID // 010000
13+
NETWORK // 100000
14+
)
15+
16+
type localityInfo struct {
17+
region string // init from workload.GetLocality().GetRegion()
18+
zone string // init from workload.GetLocality().GetZone()
19+
subZone string // init from workload.GetLocality().GetSubZone()
20+
nodeName string // init from os.Getenv("NODE_NAME"), workload.GetNode()
21+
clusterId string // init from workload.GetClusterId()
22+
network string // workload.GetNetwork()
23+
mask uint32 // mask
24+
}
25+
26+
func Valid(s string) bool {
27+
return s != ""
28+
}
29+
30+
func (l *localityInfo) Set(s string, param uint32) {
31+
if !Valid(s) {
32+
return
33+
}
34+
switch param {
35+
case REGION:
36+
l.region = s
37+
case ZONE:
38+
l.zone = s
39+
case SUBZONE:
40+
l.subZone = s
41+
case NODENAME:
42+
l.nodeName = s
43+
case CLUSTERID:
44+
l.clusterId = s
45+
case NETWORK:
46+
l.network = s
47+
}
48+
l.mask |= param
49+
}
50+
51+
func (l *localityInfo) Clear(param uint32) {
52+
l.mask &= ^param
53+
}
54+
55+
func (l *localityInfo) IsSet(param uint32) bool {
56+
return l.mask&param != 0
57+
}
58+
59+
type LocalityCache struct {
60+
LbPolicy uint32
61+
localityInfo localityInfo
62+
LbStrictIndex uint32 // for failover strict mode
63+
isLocalityInfoSet bool
64+
RoutingPreference []workloadapi.LoadBalancing_Scope
65+
isRoutingPreferenceSet bool
66+
workloadWaitQueue map[*workloadapi.Workload]struct{}
67+
}
68+
69+
func NewLocalityCache() *LocalityCache {
70+
return &LocalityCache{
71+
localityInfo: localityInfo{},
72+
isLocalityInfoSet: false,
73+
RoutingPreference: make([]workloadapi.LoadBalancing_Scope, 0),
74+
isRoutingPreferenceSet: false,
75+
workloadWaitQueue: make(map[*workloadapi.Workload]struct{}),
76+
}
77+
}
78+
79+
func (l *LocalityCache) SetLocality(nodeName, clusterId, network string, locality *workloadapi.Locality) {
80+
// notice: nodeName should set by processor or os.Getenv("NODE_NAME"),
81+
l.localityInfo.Set(nodeName, NODENAME)
82+
l.localityInfo.Set(locality.GetRegion(), REGION)
83+
l.localityInfo.Set(locality.GetSubzone(), SUBZONE)
84+
l.localityInfo.Set(locality.GetZone(), ZONE)
85+
l.localityInfo.Set(clusterId, CLUSTERID)
86+
l.localityInfo.Set(network, NETWORK)
87+
88+
l.isLocalityInfoSet = true
89+
}
90+
91+
func (l *LocalityCache) SetRoutingPreference(s []workloadapi.LoadBalancing_Scope) {
92+
// notice: s should set by lb.GetRoutingPreference()
93+
if len(s) > 0 {
94+
l.RoutingPreference = s
95+
l.LbStrictIndex = uint32(len(s))
96+
l.isRoutingPreferenceSet = true
97+
}
98+
}
99+
100+
func (l *LocalityCache) CanLocalityLB() bool {
101+
log.Debugf("isLocalityInfoSet: %#v, isRoutingPreferenceSet: %#v", l.isLocalityInfoSet, l.isRoutingPreferenceSet)
102+
return l.isLocalityInfoSet && l.isRoutingPreferenceSet
103+
}
104+
105+
func (l *LocalityCache) CalcuLocalityLBPrio(wl *workloadapi.Workload) uint32 {
106+
var rank uint32 = 0
107+
for scope := range l.RoutingPreference {
108+
switch scope {
109+
case int(workloadapi.LoadBalancing_REGION):
110+
log.Debugf("l.localityInfo.IsSet(REGION) %#v, Valid(wl.GetLocality().GetRegion()) %#v, l.localityInfo.region %#v, wl.GetLocality().GetRegion() %#v", l.localityInfo.IsSet(REGION), Valid(wl.GetLocality().GetRegion()), l.localityInfo.region, wl.GetLocality().GetRegion())
111+
if l.localityInfo.IsSet(REGION) && Valid(wl.GetLocality().GetRegion()) && l.localityInfo.region == wl.GetLocality().GetRegion() {
112+
rank++
113+
}
114+
case int(workloadapi.LoadBalancing_ZONE):
115+
log.Debugf("l.localityInfo.IsSet(ZONE) %#v, Valid(wl.GetLocality().GetZone()) %#v, l.localityInfo.zone %#v, wl.GetLocality().GetZone() %#v", l.localityInfo.IsSet(ZONE), Valid(wl.GetLocality().GetZone()), l.localityInfo.zone, wl.GetLocality().GetZone())
116+
if l.localityInfo.IsSet(ZONE) && Valid(wl.GetLocality().GetZone()) && l.localityInfo.zone == wl.GetLocality().GetZone() {
117+
rank++
118+
}
119+
case int(workloadapi.LoadBalancing_SUBZONE):
120+
log.Debugf("l.localityInfo.IsSet(SUBZONE) %#v, Valid(wl.GetLocality().GetSubzone()) %#v, l.localityInfo.subZone %#v, wl.GetLocality().GetSubzone() %#v", l.localityInfo.IsSet(SUBZONE), Valid(wl.GetLocality().GetSubzone()), l.localityInfo.subZone, wl.GetLocality().GetSubzone())
121+
if l.localityInfo.IsSet(SUBZONE) && Valid(wl.GetLocality().GetSubzone()) && l.localityInfo.subZone == wl.GetLocality().GetSubzone() {
122+
rank++
123+
}
124+
case int(workloadapi.LoadBalancing_NODE):
125+
log.Debugf("l.localityInfo.IsSet(NODENAME) %#v, Valid(wl.GetNode()) %#v, l.localityInfo.nodeName %#v, wl.GetNode() %#v", l.localityInfo.IsSet(NODENAME), Valid(wl.GetNode()), l.localityInfo.nodeName, wl.GetNode())
126+
if l.localityInfo.IsSet(NODENAME) && Valid(wl.GetNode()) && l.localityInfo.nodeName == wl.GetNode() {
127+
rank++
128+
}
129+
case int(workloadapi.LoadBalancing_NETWORK):
130+
log.Debugf("l.localityInfo.IsSet(NETWORK) %#v, Valid(wl.GetNetwork()) %#v, l.localityInfo.network %#v, wl.GetNetwork() %#v", l.localityInfo.IsSet(NETWORK), Valid(wl.GetNetwork()), l.localityInfo.network, wl.GetNetwork())
131+
if l.localityInfo.IsSet(NETWORK) && Valid(wl.GetNetwork()) && l.localityInfo.network == wl.GetNetwork() {
132+
rank++
133+
}
134+
case int(workloadapi.LoadBalancing_CLUSTER):
135+
log.Debugf("l.localityInfo.IsSet(CLUSTERID) %#v, Valid(wl.GetClusterId()) %#v, l.localityInfo.clusterId %#v, wl.GetClusterId() %#v", l.localityInfo.IsSet(CLUSTERID), Valid(wl.GetClusterId()), l.localityInfo.clusterId, wl.GetClusterId())
136+
if l.localityInfo.IsSet(CLUSTERID) && Valid(wl.GetClusterId()) && l.localityInfo.clusterId == wl.GetClusterId() {
137+
rank++
138+
}
139+
}
140+
}
141+
return rank
142+
}
143+
144+
func (l *LocalityCache) SaveToWaitQueue(wl *workloadapi.Workload) {
145+
l.workloadWaitQueue[wl] = struct{}{}
146+
}
147+
148+
func (l *LocalityCache) DelWorkloadFromWaitQueue(wl *workloadapi.Workload) {
149+
delete(l.workloadWaitQueue, wl)
150+
}
151+
152+
func (l *LocalityCache) GetFromWaitQueue() map[*workloadapi.Workload]struct{} {
153+
return l.workloadWaitQueue
154+
}

‎pkg/controller/workload/bpfcache/service.go‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,9 @@ type ServicePorts [MaxPortNum]uint32
3232
type TargetPorts [MaxPortNum]uint32
3333

3434
type ServiceValue struct {
35-
EndpointCount uint32 // endpoint count of current service
36-
LbPolicy uint32 // load balancing algorithm, currently only supports random algorithm
35+
EndpointCount [MaxPrioNum]uint32 // endpoint count of current service
36+
LbPolicy uint32 // load balancing algorithm, currently only supports random algorithm
37+
LbStrictIndex uint32
3738
ServicePort ServicePorts // ServicePort[i] and TargetPort[i] are a pair, i starts from 0 and max value is MaxPortNum-1
3839
TargetPort TargetPorts
3940
WaypointAddr [16]byte

0 commit comments

Comments
 (0)