maintain cluster active connection counter for circuit breaker #570
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -0,0 +1,178 @@ | ||||||
| /* SPDX-License-Identifier: (GPL-2.0-only OR BSD-2-Clause) */ | ||||||
| /* Copyright Authors of Kmesh */ | ||||||
|
|
||||||
| #include "bpf_log.h" | ||||||
LiZhenCheng9527 marked this conversation as resolved.
Show resolved
Hide resolved
|
||||||
| #include "kmesh_common.h" | ||||||
| #include "bpf_common.h" | ||||||
|
|
||||||
| #ifndef __KMESH_CIRCUIT_BREAKER_H__ | ||||||
| #define __KMESH_CIRCUIT_BREAKER_H__ | ||||||
|
|
||||||
| #define CLUSTER_NAME_MAX_LEN BPF_DATA_MAX_LEN | ||||||
|
|
||||||
| #pragma pack(1) | ||||||
| struct cluster_stats { | ||||||
| __u32 active_connections; | ||||||
| }; | ||||||
|
|
||||||
| struct cluster_stats_key { | ||||||
| __u64 netns_cookie; | ||||||
| __u32 cluster_id; | ||||||
| }; | ||||||
| #pragma pack() | ||||||
|
|
||||||
| struct { | ||||||
| __uint(type, BPF_MAP_TYPE_HASH); | ||||||
| __uint(key_size, sizeof(struct cluster_stats_key)); | ||||||
| __uint(value_size, sizeof(struct cluster_stats)); | ||||||
| __uint(map_flags, BPF_F_NO_PREALLOC); | ||||||
| __uint(max_entries, MAP_SIZE_OF_CLUSTER); | ||||||
| } map_of_cluster_stats SEC(".maps"); | ||||||
hzxuzhonghu marked this conversation as resolved.
Show resolved
Hide resolved
|
||||||
|
|
||||||
| struct cluster_sock_data { | ||||||
| __u32 cluster_id; | ||||||
| }; | ||||||
|
|
||||||
| struct { | ||||||
| __uint(type, BPF_MAP_TYPE_SK_STORAGE); | ||||||
| __uint(map_flags, BPF_F_NO_PREALLOC); | ||||||
| __type(key, int); | ||||||
| __type(value, struct cluster_sock_data); | ||||||
| } map_of_cluster_sock SEC(".maps"); | ||||||
|
|
||||||
| static inline void update_cluster_active_connections(const struct cluster_stats_key *key, int delta) | ||||||
hzxuzhonghu marked this conversation as resolved.
Show resolved
Hide resolved
|
||||||
| { | ||||||
| struct cluster_stats *stats = NULL; | ||||||
| if (!key) { | ||||||
| return; | ||||||
| } | ||||||
| stats = kmesh_map_lookup_elem(&map_of_cluster_stats, key); | ||||||
| if (!stats) { | ||||||
| struct cluster_stats init_value = {0}; | ||||||
| bpf_map_update_elem(&map_of_cluster_stats, key, &init_value, BPF_NOEXIST); | ||||||
| stats = kmesh_map_lookup_elem(&map_of_cluster_stats, key); | ||||||
| } | ||||||
|
|
||||||
| if (!stats) { | ||||||
| BPF_LOG(ERR, CIRCUIT_BREAKER, "failed to get cluster stats"); | ||||||
| return; | ||||||
| } | ||||||
| if (delta < 0 && -delta > stats->active_connections) { | ||||||
| BPF_LOG(ERR, CIRCUIT_BREAKER, "invalid delta update"); | ||||||
| return; | ||||||
| } | ||||||
|
|
||||||
| __sync_fetch_and_add(&stats->active_connections, delta); | ||||||
|
Member
There was a problem hiding this comment. It seems it only protect
Member
Author
There was a problem hiding this comment. since it's atomic, update will be before-or-after, only one processor will update stats->active_connections at a time.
Member
There was a problem hiding this comment. Yes, map_lookup is a memory reference, so this atomic op is concurrent safe |
||||||
|
|
||||||
| BPF_LOG( | ||||||
|
|
||||||
| DEBUG, | ||||||
| CIRCUIT_BREAKER, | ||||||
| "update existing stats(netns_cookie = %lld, cluster_id = %ld), " | ||||||
| "current active connections: %d", | ||||||
| key->netns_cookie, | ||||||
| key->cluster_id, | ||||||
| stats->active_connections); | ||||||
| } | ||||||
|
|
||||||
| static inline int on_cluster_sock_bind(ctx_buff_t *ctx, const Cluster__Cluster *cluster) | ||||||
| { | ||||||
| __u32 cluster_id = cluster->id; | ||||||
| struct cluster_stats_key key = {0}; | ||||||
| __u64 cookie = bpf_get_netns_cookie(ctx); | ||||||
| key.cluster_id = cluster_id; | ||||||
| key.netns_cookie = cookie; | ||||||
| struct cluster_stats *stats = NULL; | ||||||
| stats = kmesh_map_lookup_elem(&map_of_cluster_stats, &key); | ||||||
|
|
||||||
| if (stats != NULL) { | ||||||
| Cluster__CircuitBreakers *cbs = NULL; | ||||||
| cbs = kmesh_get_ptr_val(cluster->circuit_breakers); | ||||||
| if (cbs != NULL && stats->active_connections >= cbs->max_connections) { | ||||||
| BPF_LOG( | ||||||
| DEBUG, | ||||||
| CIRCUIT_BREAKER, | ||||||
| "Current active connections %d exceeded max connections " | ||||||
| "%d, reject connection", | ||||||
| stats->active_connections, | ||||||
| cbs->max_connections); | ||||||
| return -1; | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| BPF_LOG(DEBUG, CIRCUIT_BREAKER, "record sock bind for cluster id = %ld", cluster_id); | ||||||
|
|
||||||
| struct cluster_sock_data *data = NULL; | ||||||
| if (!ctx->sk) { | ||||||
| BPF_LOG(WARN, CIRCUIT_BREAKER, "provided sock is NULL"); | ||||||
|
Contributor
There was a problem hiding this comment.
Suggested change
Member
Author
There was a problem hiding this comment. thanks, i am busy these weeks. will modify it this weekend. |
||||||
| return 0; | ||||||
| } | ||||||
| data = bpf_sk_storage_get(&map_of_cluster_sock, ctx->sk, 0, BPF_LOCAL_STORAGE_GET_F_CREATE); | ||||||
| if (!data) { | ||||||
| BPF_LOG(ERR, CIRCUIT_BREAKER, "on_cluster_sock_bind call bpf_sk_storage_get failed"); | ||||||
| return 0; | ||||||
| } | ||||||
| data->cluster_id = cluster_id; | ||||||
| return 0; | ||||||
| } | ||||||
|
|
||||||
| static inline struct cluster_sock_data *get_cluster_sk_data(struct bpf_sock *sk) | ||||||
| { | ||||||
| struct cluster_sock_data *data = NULL; | ||||||
| if (!sk) { | ||||||
| BPF_LOG(DEBUG, CIRCUIT_BREAKER, "provided sock is NULL"); | ||||||
| return NULL; | ||||||
| } | ||||||
|
|
||||||
| data = bpf_sk_storage_get(&map_of_cluster_sock, sk, 0, 0); | ||||||
| return data; | ||||||
| } | ||||||
|
|
||||||
| static inline void on_cluster_sock_connect(struct bpf_sock_ops *ctx) | ||||||
| { | ||||||
| if (!ctx) { | ||||||
| return; | ||||||
| } | ||||||
| struct cluster_sock_data *data = get_cluster_sk_data(ctx->sk); | ||||||
| if (!data) { | ||||||
| return; | ||||||
| } | ||||||
| __u64 cookie = bpf_get_netns_cookie(ctx); | ||||||
| struct cluster_stats_key key = {0}; | ||||||
| key.netns_cookie = cookie; | ||||||
| key.cluster_id = data->cluster_id; | ||||||
| BPF_LOG( | ||||||
| DEBUG, | ||||||
| CIRCUIT_BREAKER, | ||||||
| "increase cluster active connections(netns_cookie = %lld, cluster " | ||||||
| "id = %ld)", | ||||||
| key.netns_cookie, | ||||||
| key.cluster_id); | ||||||
| update_cluster_active_connections(&key, 1); | ||||||
| BPF_LOG(DEBUG, CIRCUIT_BREAKER, "record sock connection for cluster id = %ld", data->cluster_id); | ||||||
| } | ||||||
|
|
||||||
| static inline void on_cluster_sock_close(struct bpf_sock_ops *ctx) | ||||||
| { | ||||||
| if (!ctx) { | ||||||
| return; | ||||||
| } | ||||||
| struct cluster_sock_data *data = get_cluster_sk_data(ctx->sk); | ||||||
| if (!data) { | ||||||
| return; | ||||||
| } | ||||||
| __u64 cookie = bpf_get_netns_cookie(ctx); | ||||||
| struct cluster_stats_key key = {0}; | ||||||
| key.netns_cookie = cookie; | ||||||
| key.cluster_id = data->cluster_id; | ||||||
| update_cluster_active_connections(&key, -1); | ||||||
| BPF_LOG( | ||||||
| DEBUG, | ||||||
| CIRCUIT_BREAKER, | ||||||
| "decrease cluster active connections(netns_cookie = %lld, cluster " | ||||||
| "id = %ld)", | ||||||
| key.netns_cookie, | ||||||
| key.cluster_id); | ||||||
| BPF_LOG(DEBUG, CIRCUIT_BREAKER, "record sock close for cluster id = %ld", data->cluster_id); | ||||||
| } | ||||||
|
|
||||||
| #endif | ||||||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
weired, seems the tag number is random