-
Notifications
You must be signed in to change notification settings - Fork 19
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* feat: add health service module * feat: add FunctionManager * chore: style optimize * chore: remove redundant code * chore: rename FunctionManagerBean to FunctionManagerLoader
- Loading branch information
1 parent
e952af2
commit 15b1f40
Showing
17 changed files
with
1,187 additions
and
11 deletions.
There are no files selected for viewing
This file contains 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
90 changes: 90 additions & 0 deletions
90
...console/src/main/java/org/apache/eventmesh/dashboard/console/health/CheckResultCache.java
This file contains 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,90 @@ | ||
/* | ||
* Licensed to the 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. | ||
* The 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 org.apache.eventmesh.dashboard.console.health; | ||
|
||
import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckStatus; | ||
import org.apache.eventmesh.dashboard.console.health.check.config.HealthCheckObjectConfig; | ||
|
||
import java.time.LocalDateTime; | ||
import java.util.Collections; | ||
import java.util.HashMap; | ||
import java.util.Map; | ||
import java.util.Objects; | ||
|
||
import lombok.AllArgsConstructor; | ||
import lombok.Getter; | ||
import lombok.Setter; | ||
|
||
public class CheckResultCache { | ||
|
||
private final HashMap<String, HashMap<Long, CheckResult>> cacheMap = new HashMap<>(); | ||
|
||
public void update(String type, Long typeId, HealthCheckStatus status, String resultDesc, Long latency) { | ||
HashMap<Long, CheckResult> subMap = cacheMap.get(type); | ||
if (Objects.isNull(subMap)) { | ||
subMap = new HashMap<>(); | ||
cacheMap.put(type, subMap); | ||
} | ||
CheckResult oldResult = subMap.get(typeId); | ||
String oldDesc = Objects.isNull(oldResult.getResultDesc()) ? "" : oldResult.getResultDesc() + "\n"; | ||
CheckResult result = new CheckResult(status, oldDesc + resultDesc, LocalDateTime.now(), | ||
latency, oldResult.getConfig()); | ||
subMap.put(typeId, result); | ||
} | ||
|
||
public void update(String type, Long typeId, HealthCheckStatus status, String resultDesc, Long latency, HealthCheckObjectConfig config) { | ||
HashMap<Long, CheckResult> subMap = cacheMap.get(type); | ||
if (Objects.isNull(subMap)) { | ||
subMap = new HashMap<>(); | ||
cacheMap.put(type, subMap); | ||
} | ||
CheckResult resultToUpdate = subMap.get(typeId); | ||
subMap.put(typeId, new CheckResult(status, resultDesc, LocalDateTime.now(), latency, config)); | ||
} | ||
|
||
public Map<String, HashMap<Long, CheckResult>> getCacheMap() { | ||
return Collections.unmodifiableMap(cacheMap); | ||
} | ||
|
||
@Getter | ||
@AllArgsConstructor | ||
public class CheckResult { | ||
|
||
/** | ||
* the status of a health check. | ||
* | ||
* @see HealthCheckStatus | ||
*/ | ||
@Setter | ||
private HealthCheckStatus status; | ||
/** | ||
* if not passed, this field is used to store some description. | ||
*/ | ||
private String resultDesc; | ||
/** | ||
* the time this record is inserted into memory map. | ||
*/ | ||
private LocalDateTime createTime; | ||
/** | ||
* latency of a health check, for example ping latency. | ||
*/ | ||
private Long latencyMilliSeconds; | ||
|
||
private HealthCheckObjectConfig config; | ||
} | ||
} |
140 changes: 140 additions & 0 deletions
140
...d-console/src/main/java/org/apache/eventmesh/dashboard/console/health/HealthExecutor.java
This file contains 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,140 @@ | ||
/* | ||
* Licensed to the 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. | ||
* The 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 org.apache.eventmesh.dashboard.console.health; | ||
|
||
import org.apache.eventmesh.dashboard.console.entity.health.HealthCheckResultEntity; | ||
import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckStatus; | ||
import org.apache.eventmesh.dashboard.console.enums.health.HealthCheckType; | ||
import org.apache.eventmesh.dashboard.console.health.CheckResultCache.CheckResult; | ||
import org.apache.eventmesh.dashboard.console.health.callback.HealthCheckCallback; | ||
import org.apache.eventmesh.dashboard.console.health.check.AbstractHealthCheckService; | ||
import org.apache.eventmesh.dashboard.console.service.health.HealthDataService; | ||
|
||
import java.util.ArrayList; | ||
|
||
import lombok.Getter; | ||
import lombok.Setter; | ||
import lombok.extern.slf4j.Slf4j; | ||
|
||
@Slf4j | ||
public class HealthExecutor { | ||
|
||
@Setter | ||
private HealthDataService dataService; | ||
|
||
/** | ||
* memory cache is used to store real-time health check result. | ||
*/ | ||
@Getter | ||
@Setter | ||
private CheckResultCache memoryCache; | ||
|
||
/** | ||
* execute function is where health check services work. | ||
* | ||
* @param service The health check service to be executed. | ||
*/ | ||
|
||
public void execute(AbstractHealthCheckService service) { | ||
final long startTime = System.currentTimeMillis(); | ||
//TODO: execute is called by a ScheduledThreadPoolExecutor, | ||
// when called, it should check if current service should doCheck(check service check rate can be dynamically configured). | ||
try { | ||
memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(), HealthCheckStatus.CHECKING, "", | ||
null, service.getConfig()); | ||
//The callback interface is used to pass the processing methods for checking success and failure. | ||
service.doCheck(new HealthCheckCallback() { | ||
@Override | ||
public void onSuccess() { | ||
//when the health check is successful, the result is updated to the memory cache. | ||
Long latency = System.currentTimeMillis() - startTime; | ||
HealthCheckStatus status = | ||
latency > service.getConfig().getRequestTimeoutMillis() ? HealthCheckStatus.TIMEOUT : HealthCheckStatus.PASSED; | ||
memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(), | ||
status, "health check success", latency | ||
); | ||
} | ||
|
||
@Override | ||
public void onFail(Exception e) { | ||
//when the health check fails, the result is updated to the memory cache, passing in the exception message. | ||
log.error("Health check failed for reason: {}. Service config is {}", e, service.getConfig()); | ||
memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(), | ||
HealthCheckStatus.FAILED, e.getMessage(), | ||
System.currentTimeMillis() - startTime); | ||
} | ||
}); | ||
} catch (Exception e) { | ||
log.error("Health check failed for reason: {}. Service config is {}", e, service.getConfig()); | ||
memoryCache.update(service.getConfig().getHealthCheckResourceType(), service.getConfig().getInstanceId(), HealthCheckStatus.FAILED, | ||
e.getMessage(), | ||
System.currentTimeMillis() - startTime); | ||
} | ||
} | ||
|
||
/** | ||
* this function should be called before any actual execute behaviour.<br> It will check the execution result of the last check cycle in the | ||
* memory cache, set tasks that haven't finished status to time out and update the database. | ||
*/ | ||
public void startExecute() { | ||
ArrayList<HealthCheckResultEntity> resultList = new ArrayList<>(); | ||
memoryCache.getCacheMap().forEach((type, subMap) -> { | ||
subMap.forEach((instanceId, result) -> { | ||
if (result.getStatus() == HealthCheckStatus.CHECKING) { | ||
result.setStatus(HealthCheckStatus.TIMEOUT); | ||
} | ||
addToResultList(result, resultList); | ||
}); | ||
}); | ||
if (!resultList.isEmpty()) { | ||
dataService.batchUpdateCheckResultByClusterIdAndTypeAndTypeId(resultList); | ||
} | ||
} | ||
|
||
/** | ||
* this function should be called after all actual execute behaviour.<br> It will insert the result of this check cycle into the database. At this | ||
* point the status of the tasks may be CHECKING, they will be updated on the next startExecute. | ||
*/ | ||
public void endExecute() { | ||
ArrayList<HealthCheckResultEntity> resultList = new ArrayList<>(); | ||
memoryCache.getCacheMap().forEach((type, subMap) -> { | ||
subMap.forEach((instanceId, result) -> { | ||
addToResultList(result, resultList); | ||
}); | ||
}); | ||
dataService.batchInsertHealthCheckResult(resultList); | ||
} | ||
|
||
/** | ||
* Helper function to add a CheckResult to the resultList. | ||
* | ||
* @param result memory cached result object. | ||
* @param resultList entity list to be inserted into the database. | ||
*/ | ||
private void addToResultList(CheckResult result, ArrayList<HealthCheckResultEntity> resultList) { | ||
HealthCheckResultEntity newEntity = new HealthCheckResultEntity(); | ||
newEntity.setClusterId(result.getConfig().getClusterId()); | ||
newEntity.setType(HealthCheckType.toNumber(result.getConfig().getHealthCheckResourceType())); | ||
newEntity.setTypeId(result.getConfig().getInstanceId()); | ||
newEntity.setResultDesc(result.getResultDesc()); | ||
newEntity.setStatus(result.getStatus().getNumber()); | ||
|
||
resultList.add(newEntity); | ||
} | ||
|
||
} |
Oops, something went wrong.