Skip to content

增加监听地址scope和namespace，移除部分不常用的维度 #64

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
owent merged 2 commits into from
Sep 11, 2026
Merged

增加监听地址scope和namespace，移除部分不常用的维度 #64

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d75787d
增加监听地址scope和namespace，移除部分不常用的维度
owent Sep 11, 2026
253c354
修复GCC兼容性
owent Sep 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion atframework/libatbus
Open in desktop
Submodule libatbus updated from ef1149 to d46c04
5 changes: 2 additions & 3 deletions include/atframe/atapp.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -372,11 +372,9 @@ class app {
LIBATAPP_MACRO_API void set_kind(gsl::string_view value);
LIBATAPP_MACRO_API void set_group(gsl::string_view value);

LIBATAPP_MACRO_API void set_metadata_scope(gsl::string_view value);
LIBATAPP_MACRO_API void set_metadata_name(gsl::string_view value);
LIBATAPP_MACRO_API void set_metadata_namespace_name(gsl::string_view value);
LIBATAPP_MACRO_API void set_metadata_uid(gsl::string_view value);

LIBATAPP_MACRO_API void set_metadata_service_subset(gsl::string_view value);

LIBATAPP_MACRO_API void set_metadata_label(gsl::string_view key, gsl::string_view value);

Expand Down Expand Up @@ -625,6 +623,7 @@ class app {
void print_help();

static bool match_gateway_hosts(const atapp::protocol::atapp_gateway &checked) noexcept;
bool match_gateway_scope(const atapp::protocol::atapp_gateway &checked) const noexcept;
bool match_gateway_namespace(const atapp::protocol::atapp_gateway &checked) const noexcept;
bool match_gateway_labels(const atapp::protocol::atapp_gateway &checked) const noexcept;

Expand Down
29 changes: 17 additions & 12 deletions include/atframe/atapp_conf.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,17 +54,21 @@ message atapp_area {
message atapp_gateway {
string address = 1 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

repeated string match_hosts = 11; // match atapp_configure.hostname with "In" operator
// match atapp_metadata.scope with "Equal" operator
string match_scope = 10 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

// match atapp_configure.hostname with "In" operator
repeated string match_hosts = 11 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

// Just like NetworkPolicyPeer in
// https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/api/networking/v1/generated.proto
// See also LabelSelector in
// https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/generated.proto
repeated string match_namespaces = 21 [
(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }
]; // match atapp_metadata.namespace_name with "In" operator
map<string, string> match_labels = 22
[(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }]; // match all labels
// match atapp_metadata.namespace_name with "In" operator
repeated string match_namespaces = 21 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

// match atapp_metadata.labels with "Contains" operator
map<string, string> match_labels = 22 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];
}

enum ATAPP_ACCESS_DATA_ALGORITHM_TYPE {
Expand Down Expand Up @@ -226,8 +230,12 @@ message atbus_configure {
uint64 access_token_max_number = 107;
repeated string access_tokens = 108;
repeated atapp_gateway gateways = 110;

bool overwrite_listen_path = 111 [(atapp.protocol.CONFIGURE) = { default_value: "false" }];

// Which labels will be inherited by atbus from atapp_metadata.labels
repeated string inherited_labels = 113 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

atbus_topology topology = 120;

google.protobuf.Duration first_idle_timeout = 201 [(atapp.protocol.CONFIGURE) = { default_value: "30s" }];
Expand Down Expand Up @@ -504,16 +512,13 @@ message atapp_metadata {
string kind = 2 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];
string group = 3 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

// Scope of listen addresses, usually set to cluster.metadata.uid or cluster name of Kubernetes cluster
string scope = 4 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

// kubernetes ENV: ATAPP_METADATA_NAME: valueFrom: fieldRef: fieldPath: metadata.name
string name = 11 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];
// kubernetes ENV: ATAPP_METADATA_NAMESPACE_NAME: valueFrom: fieldRef: fieldPath: metadata.namespace
string namespace_name = 12 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];
// kubernetes ENV: ATAPP_METADATA_UID: valueFrom: fieldRef: fieldPath: metadata.uid
string uid = 13 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

// `service_subset` is similar to Destination in
// https://istio.io/latest/docs/reference/config/networking/virtual-service/#Destination
string service_subset = 23 [(atframework.atapp.protocol.CONFIGURE) = { enable_expression: true }];

Comment on lines 518 to 522
// https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set
// https://github.com/kubernetes/kubernetes/blob/master/staging/src/k8s.io/cri-api/pkg/apis/runtime/v1/api.proto
Expand Down
6 changes: 5 additions & 1 deletion src/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,11 @@ add_custom_command(
"Generate ${PROJECT_LIBATAPP_GENERATED_DIR}/include/atframe/atapp_conf.pb.h and ${PROJECT_LIBATAPP_GENERATED_DIR}/src/atapp_conf.pb.cc"
)

set_property(
SOURCE ${PROJECT_LIBATAPP_SRC_LIST}
APPEND
PROPERTY OBJECT_DEPENDS "${PROJECT_LIBATAPP_GENERATED_DIR}/include/atframe/atapp_conf.pb.h")

if(NOT DEFINED LIBATAPP_MACRO_PROTOCOL_USE_DYNAMIC_LIBRARY)
if(BUILD_SHARED_LIBS OR ATFRAMEWORK_USE_DYNAMIC_LIBRARY)
if(DEFINED ATFRAMEWORK_CMAKE_TOOLSET_THIRD_PARTY_PROTOBUF_ALLOW_SHARED_LIBS)
Expand Down Expand Up @@ -207,4 +212,3 @@ install(
install(FILES "${PROJECT_LIBATAPP_GENERATED_DIR}/include/atframe/atapp_conf.pb.h"
"${PROJECT_LIBATAPP_GENERATED_DIR}/include/atframe/atapp_config.h"
DESTINATION "${ATFRAMEWORK_INSTALL_INCLUDEDIR}/atframe/")

109 changes: 83 additions & 26 deletions src/atframe/atapp.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,10 @@
#include <iomanip>
#include <iostream>
#include <limits>
#include <list>
#include <sstream>
#include <unordered_set>
#include <utility>
#include <list>
#if !(defined(ATFRAMEWORK_UTILS_THREAD_TLS_USE_PTHREAD) && ATFRAMEWORK_UTILS_THREAD_TLS_USE_PTHREAD) && \
__cplusplus >= 201103L
# include <mutex>
Expand Down Expand Up @@ -266,7 +266,8 @@ static atbus::protocol::ATBUS_COMPRESSION_LEVEL convert_atbus_configure(protocol
}
}

static void apply_atbus_configure(atbus::node::conf_t &to, const protocol::atbus_configure &from) {
static void apply_atbus_configure(atbus::node::conf_t &to, const protocol::atbus_configure &from,
const protocol::atapp_metadata &metadata) {
atbus::node::default_conf(&to);

to.upstream_address = from.proxy();
Expand All @@ -289,6 +290,46 @@ static void apply_atbus_configure(atbus::node::conf_t &to, const protocol::atbus
protobuf_to_chrono_set_duration(to.first_idle_timeout, from.first_idle_timeout());
protobuf_to_chrono_set_duration(to.ping_interval, from.ping_interval());
protobuf_to_chrono_set_duration(to.retry_interval, from.retry_interval());
protobuf_to_chrono_set_duration(to.max_retry_interval, from.reconnect_max_interval());

// network scope
to.scope = metadata.scope();
to.namespace_name = metadata.namespace_name();
to.node_labels.clear();
to.node_labels.reserve(
(std::min)(static_cast<size_t>(metadata.labels_size()), static_cast<size_t>(from.inherited_labels_size())));
std::unordered_set<std::string> inherited_labels_set;
inherited_labels_set.reserve(static_cast<size_t>(from.inherited_labels_size()));
for (const auto &label_k : from.inherited_labels()) {
inherited_labels_set.insert(label_k);
}
for (const auto &label_kv : metadata.labels()) {
if (inherited_labels_set.find(label_kv.first) != inherited_labels_set.end() && !label_kv.second.empty()) {
to.node_labels[label_kv.first] = label_kv.second;
}
}
to.gateway.clear();
to.gateway.reserve(static_cast<size_t>(from.gateways_size()));
for (const auto &gw : from.gateways()) {
if (gw.address().empty()) {
continue;
}
to.gateway.push_back(atbus::node::gateway_t());
auto &gw_cfg = to.gateway.back();
gw_cfg.address = gw.address();
gw_cfg.match_scope = gw.match_scope();
gw_cfg.match_hosts.reserve(static_cast<size_t>(gw.match_hosts().size()));
gw_cfg.match_hosts = {gw.match_hosts().begin(), gw.match_hosts().end()};
gw_cfg.match_namespaces.reserve(static_cast<size_t>(gw.match_namespaces().size()));
gw_cfg.match_namespaces = {gw.match_namespaces().begin(), gw.match_namespaces().end()};
Comment on lines +319 to +324
gw_cfg.match_labels.reserve(
(std::min)(static_cast<size_t>(gw.match_labels().size()), static_cast<size_t>(from.inherited_labels_size())));
for (const auto &label_kv : gw.match_labels()) {
if (inherited_labels_set.find(label_kv.first) != inherited_labels_set.end() && !label_kv.second.empty()) {
gw_cfg.match_labels[label_kv.first] = label_kv.second;
}
}
}

to.fault_tolerant = static_cast<size_t>(from.fault_tolerant());
to.message_size = static_cast<size_t>(from.message_size());
Expand Down Expand Up @@ -1175,6 +1216,9 @@ LIBATAPP_MACRO_API int app::reload() {
bus_node_->reload_compression(gsl::span<const atbus::protocol::ATBUS_COMPRESSION_ALGORITHM_TYPE>(
conf_.bus_conf.compression_allow_algorithms),
conf_.bus_conf.compression_level);
bus_node_->reload_self_endpoint(
conf_.bus_conf.scope, conf_.bus_conf.namespace_name, conf_.bus_conf.node_labels,
gsl::span<const atbus::node::gateway_t>{conf_.bus_conf.gateway.data(), conf_.bus_conf.gateway.size()});
}
if (atbus_connector_) {
atbus_connector_->reload();
Expand Down Expand Up @@ -1687,53 +1731,41 @@ LIBATAPP_MACRO_API void app::set_group(gsl::string_view value) {
conf_.metadata.set_group(value.data(), value.size());
}

LIBATAPP_MACRO_API void app::set_metadata_name(gsl::string_view value) {
if (gsl::string_view(conf_.metadata.name().data(), conf_.metadata.name().size()) == value) {
return;
}

if (internal_module_service_discovery_) {
internal_module_service_discovery_->set_maybe_update_keepalive_discovery_metadata();
}

conf_.metadata.set_name(value.data(), value.size());
conf_.runtime_pod_stateful_index = static_cast<int32_t>(atapp_pod_stateful_index::kUnset);
}

LIBATAPP_MACRO_API void app::set_metadata_namespace_name(gsl::string_view value) {
if (gsl::string_view(conf_.metadata.namespace_name().c_str(), conf_.metadata.namespace_name().size()) == value) {
LIBATAPP_MACRO_API void app::set_metadata_scope(gsl::string_view value) {
if (gsl::string_view(conf_.metadata.scope().c_str(), conf_.metadata.scope().size()) == value) {
return;
}

if (internal_module_service_discovery_) {
internal_module_service_discovery_->set_maybe_update_keepalive_discovery_metadata();
}

conf_.metadata.set_namespace_name(value.data(), value.size());
conf_.metadata.set_scope(value.data(), value.size());
}

LIBATAPP_MACRO_API void app::set_metadata_uid(gsl::string_view value) {
if (gsl::string_view(conf_.metadata.uid().c_str(), conf_.metadata.uid().size()) == value) {
LIBATAPP_MACRO_API void app::set_metadata_name(gsl::string_view value) {
if (gsl::string_view(conf_.metadata.name().data(), conf_.metadata.name().size()) == value) {
return;
}

if (internal_module_service_discovery_) {
internal_module_service_discovery_->set_maybe_update_keepalive_discovery_metadata();
}

conf_.metadata.set_uid(value.data(), value.size());
conf_.metadata.set_name(value.data(), value.size());
conf_.runtime_pod_stateful_index = static_cast<int32_t>(atapp_pod_stateful_index::kUnset);
}

LIBATAPP_MACRO_API void app::set_metadata_service_subset(gsl::string_view value) {
if (gsl::string_view(conf_.metadata.service_subset().c_str(), conf_.metadata.service_subset().size()) == value) {
LIBATAPP_MACRO_API void app::set_metadata_namespace_name(gsl::string_view value) {
if (gsl::string_view(conf_.metadata.namespace_name().c_str(), conf_.metadata.namespace_name().size()) == value) {
return;
}

if (internal_module_service_discovery_) {
internal_module_service_discovery_->set_maybe_update_keepalive_discovery_metadata();
}

conf_.metadata.set_service_subset(value.data(), value.size());
conf_.metadata.set_namespace_name(value.data(), value.size());
}

LIBATAPP_MACRO_API void app::set_metadata_label(gsl::string_view key, gsl::string_view value) {
Expand Down Expand Up @@ -2651,11 +2683,15 @@ LIBATAPP_MACRO_API bool app::match_gateway(const atapp::protocol::atapp_gateway
return false;
}

if (!checked.match_scope().empty() && !match_gateway_scope(checked)) {
return false;
}

if (checked.match_hosts_size() > 0 && !match_gateway_hosts(checked)) {
return false;
}

if (checked.match_namespaces_size() && !match_gateway_namespace(checked)) {
if (checked.match_namespaces_size() > 0 && !match_gateway_namespace(checked)) {
return false;
}

Expand Down Expand Up @@ -2991,7 +3027,7 @@ int app::apply_configure() {
}

// atbus configure
apply_atbus_configure(conf_.bus_conf, conf_.origin.bus());
apply_atbus_configure(conf_.bus_conf, conf_.origin.bus(), conf_.metadata);

// windows minidump configure
setup_windows_minidump(conf_.origin.debug(), conf_.origin.name());
Expand Down Expand Up @@ -3711,6 +3747,17 @@ LIBATAPP_MACRO_API void app::trigger_event_on_discovery_event(etcd_discovery_act
FWLOGINFO("app {}({}, type={}:{}) got a PUT discovery event({}({}, type={}:{}))", get_app_name(), get_app_id(),
get_type_id(), get_type_name(), discovery_info.name(), discovery_info.id(), discovery_info.type_id(),
discovery_info.type_name());
if (0 != discovery_info.id()) {
auto *ep = get_endpoint(discovery_info.id());
if (ep != nullptr) {
ep->update_discovery(node);
}
} else {
auto *ep = get_endpoint(discovery_info.name());
if (ep != nullptr) {
ep->update_discovery(node);
}
}
} else {
FWLOGINFO("app {}({}, type={}:{}) got a DELETE discovery event({}({}, type={}:{})", get_app_name(), get_app_id(),
get_type_id(), get_type_name(), discovery_info.name(), discovery_info.id(), discovery_info.type_id(),
Expand Down Expand Up @@ -4326,6 +4373,14 @@ void app::print_help() {
}
}

bool app::match_gateway_scope(const atapp::protocol::atapp_gateway &checked) const noexcept {
if (checked.match_scope().empty()) {
return true;
}

return checked.match_scope() == get_metadata().scope();
}

bool app::match_gateway_hosts(const atapp::protocol::atapp_gateway &checked) noexcept {
bool has_matched_value = false;
bool has_valid_conf = false;
Expand Down Expand Up @@ -5450,6 +5505,8 @@ int app::send_last_command(ev_loop_t *ev_loop) {
bool is_sync_channel = false;
atbus::channel::channel_address_t use_addr;

// 发送指令的节点必须加载和目标节点相同的配置，并且部署在同一机器上。所以scope、namespace、label等都是相同的
// 这里就不用再根据这些判定一次连接可达性
for (int i = 0; i < conf_.origin.bus().listen_size(); ++i) {
atbus::channel::channel_address_t parsed_addr;
make_address(conf_.origin.bus().listen(i), parsed_addr);
Expand Down
Loading
Loading