Skip to content
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

Jump to bottom

Pick self monitor: merge 1.8.x #1773

Merged
merged 19 commits into from
Oct 11, 2024
Merged

Pick self monitor: merge 1.8.x #1773

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
e381210
feat: add ReentrantMetricsRecord and PluginMetricManager for c++ self…
Takuka0311 Sep 20, 2024
90961e6
feat: add loong collector monitor
Takuka0311 Sep 20, 2024
ab24e66
refactor: add variable label value to self metric apis
Takuka0311 Sep 23, 2024
79162ca
feat: export go metrics to cpp
Takuka0311 Sep 23, 2024
b258a70
feat: Upgrade PluginID to PluginMeta (CPP part)
Takuka0311 Sep 23, 2024
87031f0
feat: Upgrade PluginID to PluginMeta (Golang part)
Takuka0311 Sep 23, 2024
b107f8d
Merge remote-tracking branch 'alibaba/2.1' into add-self-monitor
Takuka0311 Sep 23, 2024
faac375
Unified PluginID Sequencing for C++ and Go Components
Takuka0311 Sep 23, 2024
67db763
fix lint and e2e
Takuka0311 Sep 24, 2024
6346c8d
fix ut
Takuka0311 Sep 24, 2024
a13856e
fix e2e
Takuka0311 Sep 26, 2024
dc28b08
fix bug
Takuka0311 Sep 29, 2024
472a3c6
test
Takuka0311 Sep 29, 2024
cb1112a
del test
Takuka0311 Sep 29, 2024
358cd7e
del e2e
Takuka0311 Sep 29, 2024
49d6e0a
fix lint
Takuka0311 Sep 30, 2024
e6d95a5
fix e2e
Takuka0311 Oct 10, 2024
ac4fcb2
fix e2e
Takuka0311 Oct 10, 2024
562335e
fix lint
Takuka0311 Oct 10, 2024
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
76 changes: 37 additions & 39 deletions core/config/Config.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,8 +104,7 @@ bool Config::Parse() {
string key, errorMsg;
const Json::Value* itr = nullptr;
LogtailAlarm& alarm = *LogtailAlarm::GetInstance();
#ifdef __ENTERPRISE__
// to send alarm, project, logstore and region should be extracted first.
// to send alarm and init MetricsRecord, project, logstore and region should be extracted first.
key = "flushers";
itr = mDetail->find(key.c_str(), key.c_str() + key.size());
if (itr && itr->isArray()) {
Expand All @@ -122,7 +121,6 @@ bool Config::Parse() {
}
}
}
#endif

if (!GetOptionalUIntParam(*mDetail, "createTime", mCreateTime, errorMsg)) {
PARAM_WARNING_DEFAULT(sLogger, alarm, errorMsg, mCreateTime, noModule, mName, mProject, mLogstore, mRegion);
Expand Down Expand Up @@ -198,19 +196,19 @@ bool Config::Parse() {
mLogstore,
mRegion);
}
const string pluginName = it->asString();
const string pluginType = it->asString();
if (i == 0) {
if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
mHasGoInput = true;
} else if (PluginRegistry::GetInstance()->IsValidNativeInputPlugin(pluginName)) {
} else if (PluginRegistry::GetInstance()->IsValidNativeInputPlugin(pluginType)) {
mHasNativeInput = true;
} else {
PARAM_ERROR_RETURN(
sLogger, alarm, "unsupported input plugin", pluginName, mName, mProject, mLogstore, mRegion);
sLogger, alarm, "unsupported input plugin", pluginType, mName, mProject, mLogstore, mRegion);
}
} else {
if (mHasGoInput) {
if (PluginRegistry::GetInstance()->IsValidNativeInputPlugin(pluginName)) {
if (PluginRegistry::GetInstance()->IsValidNativeInputPlugin(pluginType)) {
PARAM_ERROR_RETURN(sLogger,
alarm,
"native and extended input plugins coexist",
Expand All @@ -219,12 +217,12 @@ bool Config::Parse() {
mProject,
mLogstore,
mRegion);
} else if (!PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
} else if (!PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
PARAM_ERROR_RETURN(
sLogger, alarm, "unsupported input plugin", pluginName, mName, mProject, mLogstore, mRegion);
sLogger, alarm, "unsupported input plugin", pluginType, mName, mProject, mLogstore, mRegion);
}
} else {
if (PluginRegistry::GetInstance()->IsValidNativeInputPlugin(pluginName)) {
if (PluginRegistry::GetInstance()->IsValidNativeInputPlugin(pluginType)) {
PARAM_ERROR_RETURN(sLogger,
alarm,
"more than 1 native input plugin is given",
Expand All @@ -233,7 +231,7 @@ bool Config::Parse() {
mProject,
mLogstore,
mRegion);
} else if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
} else if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
PARAM_ERROR_RETURN(sLogger,
alarm,
"native and extended input plugins coexist",
Expand All @@ -244,15 +242,15 @@ bool Config::Parse() {
mRegion);
} else {
PARAM_ERROR_RETURN(
sLogger, alarm, "unsupported input plugin", pluginName, mName, mProject, mLogstore, mRegion);
sLogger, alarm, "unsupported input plugin", pluginType, mName, mProject, mLogstore, mRegion);
}
}
}
mInputs.push_back(&plugin);
if (pluginName == "input_observer_network") {
if (pluginType == "input_observer_network") {
hasObserverInput = true;
#ifdef __ENTERPRISE__
} else if (pluginName == "input_stream") {
} else if (pluginType == "input_stream") {
if (!AppConfig::GetInstance()->GetOpenStreamLog()) {
PARAM_ERROR_RETURN(
sLogger, alarm, "stream log is not enabled", noModule, mName, mProject, mLogstore, mRegion);
Expand Down Expand Up @@ -322,9 +320,9 @@ bool Config::Parse() {
mLogstore,
mRegion);
}
const string pluginName = it->asString();
const string pluginType = it->asString();
if (mHasGoInput) {
if (PluginRegistry::GetInstance()->IsValidNativeProcessorPlugin(pluginName)) {
if (PluginRegistry::GetInstance()->IsValidNativeProcessorPlugin(pluginType)) {
PARAM_ERROR_RETURN(sLogger,
alarm,
"native processor plugins coexist with extended input plugins",
Expand All @@ -333,33 +331,33 @@ bool Config::Parse() {
mProject,
mLogstore,
mRegion);
} else if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
} else if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
mHasGoProcessor = true;
} else {
PARAM_ERROR_RETURN(sLogger,
alarm,
"unsupported processor plugin",
pluginName,
pluginType,
mName,
mProject,
mLogstore,
mRegion);
}
} else {
if (isCurrentPluginNative) {
if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
isCurrentPluginNative = false;
mHasGoProcessor = true;
} else if (!PluginRegistry::GetInstance()->IsValidNativeProcessorPlugin(pluginName)) {
} else if (!PluginRegistry::GetInstance()->IsValidNativeProcessorPlugin(pluginType)) {
PARAM_ERROR_RETURN(sLogger,
alarm,
"unsupported processor plugin",
pluginName,
pluginType,
mName,
mProject,
mLogstore,
mRegion);
} else if (pluginName == "processor_spl") {
} else if (pluginType == "processor_spl") {
isSPL = true;
if (i != 0 || itr->size() != 1) {
PARAM_ERROR_RETURN(sLogger,
Expand All @@ -385,22 +383,22 @@ bool Config::Parse() {
mHasNativeProcessor = true;
}
} else {
if (PluginRegistry::GetInstance()->IsValidNativeProcessorPlugin(pluginName)) {
if (PluginRegistry::GetInstance()->IsValidNativeProcessorPlugin(pluginType)) {
PARAM_ERROR_RETURN(sLogger,
alarm,
"native processor plugin comes after extended processor plugin",
pluginName,
pluginType,
mName,
mProject,
mLogstore,
mRegion);
} else if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
} else if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
mHasGoProcessor = true;
} else {
PARAM_ERROR_RETURN(sLogger,
alarm,
"unsupported processor plugin",
pluginName,
pluginType,
mName,
mProject,
mLogstore,
Expand All @@ -410,7 +408,7 @@ bool Config::Parse() {
}
mProcessors.push_back(&plugin);
if (i == 0) {
if (pluginName == "processor_parse_json_native" || pluginName == "processor_json") {
if (pluginType == "processor_parse_json_native" || pluginType == "processor_json") {
mIsFirstProcessorJson = true;
}
}
Expand Down Expand Up @@ -471,10 +469,10 @@ bool Config::Parse() {
mLogstore,
mRegion);
}
const string pluginName = it->asString();
if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
const string pluginType = it->asString();
if (PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
mHasGoFlusher = true;
} else if (PluginRegistry::GetInstance()->IsValidNativeFlusherPlugin(pluginName)) {
} else if (PluginRegistry::GetInstance()->IsValidNativeFlusherPlugin(pluginType)) {
mHasNativeFlusher = true;
// processor spl could change pipelineEventGroup tags and affect the merge logic of aggregator
// so force specify the MergeType as logstore, loggroup will be merged into a List.
Expand All @@ -484,10 +482,10 @@ bool Config::Parse() {
}
} else {
PARAM_ERROR_RETURN(
sLogger, alarm, "unsupported flusher plugin", pluginName, mName, mProject, mLogstore, mRegion);
sLogger, alarm, "unsupported flusher plugin", pluginType, mName, mProject, mLogstore, mRegion);
}
#ifdef __ENTERPRISE__
if (hasStreamInput && pluginName != "flusher_sls") {
if (hasStreamInput && pluginType != "flusher_sls") {
PARAM_ERROR_RETURN(sLogger,
alarm,
"flusher plugins other than flusher_sls coexist with input_stream",
Expand Down Expand Up @@ -562,10 +560,10 @@ bool Config::Parse() {
mLogstore,
mRegion);
}
const string pluginName = it->asString();
if (!PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
const string pluginType = it->asString();
if (!PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
PARAM_ERROR_RETURN(
sLogger, alarm, "unsupported aggregator plugin", pluginName, mName, mProject, mLogstore, mRegion);
sLogger, alarm, "unsupported aggregator plugin", pluginType, mName, mProject, mLogstore, mRegion);
}
mAggregators.push_back(&plugin);
}
Expand Down Expand Up @@ -628,10 +626,10 @@ bool Config::Parse() {
mLogstore,
mRegion);
}
const string pluginName = it->asString();
if (!PluginRegistry::GetInstance()->IsValidGoPlugin(pluginName)) {
const string pluginType = it->asString();
if (!PluginRegistry::GetInstance()->IsValidGoPlugin(pluginType)) {
PARAM_ERROR_RETURN(
sLogger, alarm, "unsupported extension plugin", pluginName, mName, mProject, mLogstore, mRegion);
sLogger, alarm, "unsupported extension plugin", pluginType, mName, mProject, mLogstore, mRegion);
}
mExtensions.push_back(&plugin);
}
Expand Down
2 changes: 1 addition & 1 deletion core/config/provider/CommonConfigProvider.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ void CommonConfigProvider::Stop() {
void CommonConfigProvider::CheckUpdateThread() {
LOG_INFO(sLogger, ("common config provider", "started"));
usleep((rand() % 10) * 100 * 1000);
int32_t lastCheckTime = 0;
int32_t lastCheckTime = time(NULL);
unique_lock<mutex> lock(mThreadRunningMux);
while (mIsThreadRunning) {
int32_t curTime = time(NULL);
Expand Down
1 change: 1 addition & 0 deletions core/container_manager/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,3 +19,4 @@ file(GLOB LIB_SOURCE_FILES *.cpp *.h)
append_source_files(LIB_SOURCE_FILES)
add_library(${PROJECT_NAME} STATIC ${LIB_SOURCE_FILES})
target_link_libraries(${PROJECT_NAME} common)
target_link_libraries(${PROJECT_NAME} pipeline)
Loading
Loading