Skip to content
This repository was archived by the owner on Nov 21, 2025. It is now read-only.

Add sync fitting #773

Open
RayCiel wants to merge 8 commits into
base: master
Choose a base branch
from
Open

Add sync fitting #773

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8eecc6d
add sync fitting
Apr 26, 2021
67e2847
merge commit and signed off
Apr 30, 2021
b00dc2b
modify
May 10, 2021
94313ad
merge & remove hackings
May 19, 2021
d0e4d9b
optimize
Jun 1, 2021
ce2475c
add redeploy recommend
Jul 8, 2021
730fb73
fix bugs
Jul 9, 2021
dec9c8c
delete useless annotations
Jul 9, 2021
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
Binary file added aggregator/commons-math3-3.6.1-javadoc.jar
View file
Open in desktop
Binary file not shown.
Binary file added aggregator/commons-math3-3.6.1-sources.jar
View file
Open in desktop
Binary file not shown.
Binary file added aggregator/commons-math3-3.6.1-test-sources.jar
View file
Open in desktop
Binary file not shown.
Binary file added aggregator/commons-math3-3.6.1-tests.jar
View file
Open in desktop
Binary file not shown.
Binary file added aggregator/commons-math3-3.6.1-tools.jar
View file
Open in desktop
Binary file not shown.
Binary file added aggregator/commons-math3-3.6.1.jar
View file
Open in desktop
Binary file not shown.
Binary file added aggregator/javacsv.jar
View file
Open in desktop
Binary file not shown.
5 changes: 4 additions & 1 deletion aggregator/src/main/java/com/vmware/flowgate/aggregator/config/ServiceKeyConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@ public ServiceKeyConfig() {
}

public String getServiceKey() {
return serviceKey;

@yixingjia yixingjia May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please check those code before create the pr.

//String test = FlowgateConstant.serviceKey;
//this.serviceKey = System.getenv(FlowgateConstant.serviceKey);
return "loul";
//return serviceKey;
}
}
14 changes: 13 additions & 1 deletion ...r/src/main/java/com/vmware/flowgate/aggregator/scheduler/job/AggregatorJobDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
}
long execount = Long.valueOf(execountString);
//will execute weekly?

if (execount++ % 168 == 0) {
try {
EventMessage eventMessage = EventMessageUtil.createEventMessage(EventType.Aggregator,
Expand Down Expand Up @@ -76,7 +77,18 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
}catch(IOException e) {
logger.error("Failed to Send sync summary data command", e);
}
}else {
}else if(execount % 288 == 0){
try {
EventMessage eventMessage = EventMessageUtil.createEventMessage(EventType.Aggregator,
EventMessageUtil.SYNC_FITTING, "");
String jobmessage = EventMessageUtil.convertEventMessageAsString(eventMessage);
publisher.publish(EventMessageUtil.AggregatorTopic, jobmessage);
logger.info("Send sync fitting command");
// System.out.println("Send sync fitting command");
}catch(IOException e) {
logger.error("Failed to Send sync fitting command", e);
}
}else {
//will execute hourly?
try {
EventMessage eventMessage = EventMessageUtil.createEventMessage(EventType.Aggregator,
Expand Down
272 changes: 267 additions & 5 deletions aggregator/src/main/java/com/vmware/flowgate/aggregator/scheduler/job/AggregatorService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,13 @@
*/
package com.vmware.flowgate.aggregator.scheduler.job;

import java.io.FileNotFoundException;
import java.io.IOException;
import com.csvreader.CsvReader;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
Expand All @@ -21,7 +25,11 @@
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import org.apache.commons.math3.fitting.PolynomialCurveFitter;
import org.apache.commons.math3.fitting.WeightedObservedPoints;
import org.apache.commons.math3.util.Pair;
import com.fasterxml.jackson.core.JsonProcessingException;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.vmware.flowgate.aggregator.config.ServiceKeyConfig;
Expand All @@ -31,6 +39,7 @@
import com.vmware.flowgate.common.MetricName;
import com.vmware.flowgate.common.model.Asset;
import com.vmware.flowgate.common.model.FacilitySoftwareConfig;
import com.vmware.flowgate.common.model.MetricData;
import com.vmware.flowgate.common.model.FacilitySoftwareConfig.SoftwareType;
import com.vmware.flowgate.common.model.SDDCSoftwareConfig;
import com.vmware.flowgate.common.model.ServerMapping;
Expand All @@ -40,6 +49,9 @@
import com.vmware.flowgate.common.model.redis.message.EventUser;
import com.vmware.flowgate.common.model.redis.message.impl.EventMessageUtil;

import io.netty.handler.codec.string.StringDecoder;


@Service
public class AggregatorService implements AsyncService {

Expand All @@ -65,10 +77,10 @@ public void executeAsync(EventMessage message) {
logger.warn("Drop none aggregator message " + message.getType());
return;
}
logger.info(message.getContent());
//logger.info("message: " + message.getContent());

@yixingjia yixingjia May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

? why remove the log info?

Set<EventUser> users = message.getTarget().getUsers();
for (EventUser command : users) {
logger.info(command.getId());
//logger.info(command.getId());
switch (command.getId()) {
case EventMessageUtil.FullMappingCommand:
mergeServerMapping();
Expand All @@ -92,6 +104,19 @@ public void executeAsync(EventMessage message) {
case EventMessageUtil.CleanRealtimeData:
cleanRealtimeData();
break;
case EventMessageUtil.SYNC_FITTING:
List<List<Double>> results = syncFitting(false);

@yixingjia yixingjia May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

?

for (int i = 0; i < results.size(); i++)
{
String res = "Syncfitting results-" + String.valueOf(i) + " :";
for (Double param : (List<Double>)results.get(i))
{
res += " ";
res += String.valueOf(param);
}
logger.info(res);
}
break;
case EventMessageUtil.AggregateAndCleanPowerIQPDU:
aggregateAndCleanPDUFromPowerIQ();
break;
Expand All @@ -103,7 +128,247 @@ public void executeAsync(EventMessage message) {
}
}
}

public List<Double> doubleToList(double[] arr_double) {

@yixingjia yixingjia May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you move those math related code into a separate file?

List<Double> list = new ArrayList<Double>();
int num = arr_double.length;
Double [] arr_Double = new Double[num];
for(int i = 0; i < num; i++) {
arr_Double[i] = arr_double[i];
}
list = Arrays.asList(arr_Double);
return list;
}



public List<Pair<Double, Double>> MAD(List<Pair<Double, Double>> dataset, double n) {
List<Pair<Double, Double>> new_data = new ArrayList<>();
List<Double> CPU = new ArrayList<>();
for (int i = 0; i < dataset.size(); i++) {
CPU.add(dataset.get(i).getFirst());
}
double median = median(CPU);
List<Double> deviations = new ArrayList<>();
for (int i = 0; i < dataset.size(); i++) {
deviations.add(i, Math.abs(dataset.get(i).getFirst() - median));
}
double mad = median(deviations);
for (int i = 0; i < dataset.size(); i++)
{
if (Math.abs(dataset.get(i).getFirst() - median) <= n * mad)
{
new_data.add(dataset.get(i));
}
}
return new_data;
}
public static int partition(List<Double> nums, int start, int end){
int left = start;
int right = end;
double pivot = nums.get(left);
while (left < right){
while (left < right && nums.get(right) >= pivot) {
right--;
}
if (left < right) {
nums.set(left, nums.get(right));
left++;
}
while (left < right && nums.get(left) <= pivot){
left++;
}
if (left < right) {
nums.set(right, nums.get(left));
right--;
}
}
nums.set(left, pivot);
return left;
}


public static double median(List<Double> nums){
if (nums.size() == 0)
return 0;
int start = 0;
int end = nums.size() - 1;
int index = partition(nums, start, end);
if (nums.size() % 2 == 0){
while (index != nums.size() / 2 - 1){
if (index > nums.size() / 2 - 1){
index = partition(nums, start, index - 1);
} else {
index=partition(nums, index+1, end);
}
}
} else {
while (index != nums.size() / 2) {
if (index > nums.size() / 2) {
index = partition(nums, start, index - 1);
} else {
index = partition(nums, index + 1, end);
}
}
}
return nums.get(index);
}


public List<Double> doFitting(List<MetricData> MetricDatas) {

List<Double> CPU = new ArrayList<>();
List<Double> power = new ArrayList<>();
List<Pair<Long, Double>> raw_CPU_list = new ArrayList<>();
List<Pair<Long, Double>> raw_power_list = new ArrayList<>();

for (int i = 0; i < MetricDatas.size(); i++) {
if (MetricDatas.get(i).getMetricName() == "CpuUsage") {
raw_CPU_list.add(new Pair<Long, Double> (MetricDatas.get(i).getTimeStamp(), MetricDatas.get(i).getValueNum()));
}
else if (MetricDatas.get(i).getMetricName() == "Power") {
raw_power_list.add(new Pair<Long, Double> (MetricDatas.get(i).getTimeStamp(), MetricDatas.get(i).getValueNum()));
}
}
Pair<Long, Double>[] raw_CPU = new Pair[raw_CPU_list.size()];
Pair<Long, Double>[] raw_power = new Pair[raw_power_list.size()];
for (int i = 0; i < raw_CPU_list.size(); i++) {
raw_CPU[i] = raw_CPU_list.get(i);
}
for (int i = 0; i < raw_power_list.size(); i++) {
raw_power[i] = raw_power_list.get(i);
}

//Sort the pair list according the time in reverse order.
Arrays.sort(raw_CPU, new Comparator<Pair<Long, Double>>() {
@Override
public int compare(Pair<Long, Double> o1, Pair<Long, Double> o2) {
if(o1.getFirst()==o2.getFirst()){
return 0;
}else if (o1.getFirst() > o2.getFirst()){
return -1;
}
else return 1;
}
});
Arrays.sort(raw_power, new Comparator<Pair<Long, Double>>() {
@Override
public int compare(Pair<Long, Double> o1, Pair<Long, Double> o2) {
if(o1.getFirst()==o2.getFirst()){
return 0;
}else if (o1.getFirst() > o2.getFirst()){
return -1;
}
else return 1;
}
});

int idx_CPU = 0, idx_power = 0;
List<Pair<Double, Double>> raw_data = new ArrayList<>();
while (idx_CPU < raw_CPU.length && idx_power < raw_power.length) {
if (raw_CPU[idx_CPU].getFirst().compareTo(raw_power[idx_power].getFirst()) == 0) {
raw_data.add(new Pair<Double, Double>(raw_CPU[idx_CPU].getSecond(), raw_power[idx_power].getSecond()));
idx_CPU +=1;
idx_power +=1;
}
else if (raw_CPU[idx_CPU].getFirst().compareTo(raw_power[idx_power].getFirst()) == 1) {
idx_CPU += 1;
}
else if (raw_CPU[idx_CPU].getFirst().compareTo(raw_power[idx_power].getFirst()) == -1) {
idx_power += 1;
}
}


List<Pair<Double, Double>> new_data = new ArrayList<>();
WeightedObservedPoints points = new WeightedObservedPoints();
while (raw_data.size() != 0) {
int count = 0;
for (int i = 1; i < raw_data.size(); i++) {

if (raw_data.get(i).getSecond() >= raw_data.get(i-1).getSecond() + 1)
break;
count += 1;
if (count > 0) {
List<Pair<Double, Double>> tmp = raw_data.subList(0, count + 1);
new_data.addAll(MAD(tmp, 1.5));
}
raw_data = raw_data.subList(count + 1, raw_data.size());
}
}
for(int i = 0; i < new_data.size(); i++)
{
points.add(new_data.get(i).getFirst(), new_data.get(i).getSecond());
}
int degree = 4;
PolynomialCurveFitter fitter = PolynomialCurveFitter.create(degree);
double[] result = fitter.fit(points.toList());
List<Double> fitting_result = doubleToList(result);



return fitting_result;
}

public List<List<Double>> syncFitting(boolean ifTest) {

//
List<MetricData> MetricDatas;
List<List<Double>> results = new ArrayList<>();
if (ifTest) {
try {
MetricDatas = new ArrayList<>();

CsvReader csvReader = new CsvReader("testData.csv");

@yixingjia yixingjia May 13, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

you should change to api for the when you submit pr

boolean re = csvReader.readHeaders();
int n = 0;
while (csvReader.readRecord()) {
String rawRecord = csvReader.getRawRecord();
String[] line = rawRecord.split(",");
MetricData cpu = new MetricData();
cpu.setMetricName("CpuUsage");
cpu.setValueNum(Double.valueOf(line[0]));
cpu.setTimeStamp(n);
MetricDatas.add(cpu);
MetricData power = new MetricData();
power.setMetricName("Power");
power.setValueNum(Double.valueOf(line[1]));
power.setTimeStamp(n);
MetricDatas.add(power);
n+=1;
}
List<Double> fitting_result = doFitting(MetricDatas);

results.add(fitting_result);

} catch (FileNotFoundException e) {
throw new RuntimeException("file not found");
} catch (IOException e) {
throw new RuntimeException(e.getMessage());
}
}
else {
long One_day = 86405000;
restClient.setServiceKey(serviceKeyConfig.getServiceKey());
MetricData[] raw_MetricDatas = null;
Asset[] servers = restClient.getMappedAsset(AssetCategory.Server).getBody();
for (int i = 0; i < servers.length; i++)
{
String assetId = servers[i].getId();
raw_MetricDatas = restClient.getServerRealtimeDataByServerID(assetId, System.currentTimeMillis(), One_day).getBody();
MetricDatas = Arrays.asList(raw_MetricDatas);
List<Double> fitting_result = doFitting(MetricDatas);
results.add(fitting_result);
Asset asset = new Asset();
asset.setFittingResults(fitting_result);
restClient.saveAssets(asset);

}

}
return results;
}

public void syncSummaryData() {
restClient.setServiceKey(serviceKeyConfig.getServiceKey());
restClient.getSystemSummary(false);
Expand Down Expand Up @@ -372,9 +637,6 @@ public Map<String,Map<String,String>> generatePduformulaForServer (List<String>
metricNameAndIdMap.put(MetricName.SERVER_USED_PDU_OUTLET_CURRENT, pduAssetId);
metricNameAndIdMap.put(MetricName.SERVER_USED_PDU_OUTLET_POWER, pduAssetId);
metricNameAndIdMap.put(MetricName.SERVER_VOLTAGE, pduAssetId);
metricNameAndIdMap.put(MetricName.SERVER_USED_PDU_OUTLET_VOLTAGE, pduAssetId);
metricNameAndIdMap.put(MetricName.SERVER_CONNECTED_PDU_POWER_LOAD, pduAssetId);
metricNameAndIdMap.put(MetricName.SERVER_CONNECTED_PDU_CURRENT_LOAD, pduAssetId);
pduFormula.put(pduAssetId, metricNameAndIdMap);
}
return pduFormula;
Expand Down
3 changes: 2 additions & 1 deletion ...ator/src/main/java/com/vmware/flowgate/aggregator/scheduler/job/VCenterJobDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
//every 12 hour we will trigger a sync host metadata job.
//every 1 day we will trigger a sync CustomerAttrsData job.
//every 10 days we will trigger a sync CustomAttributes job.
//every 1 day we will trigger a fitting job.

restClient.setServiceKey(serviceKeyConfig.getServiceKey());
SDDCSoftwareConfig[] vcServers = restClient.getVCServers().getBody();
Expand Down Expand Up @@ -110,7 +111,7 @@ public void execute(JobExecutionContext context) throws JobExecutionException {
generateSDDCMessageListByType(EventMessageUtil.VCENTER_SyncCustomerAttrs,
vcServersActiveArray));
}

publisher.publish(EventMessageUtil.VCTopic,
EventMessageUtil.generateSDDCNotifyMessage(EventType.VCenter));
} catch (IOException e) {
Expand Down
2 changes: 1 addition & 1 deletion aggregator/src/main/resources/application.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,6 @@ spring.data.redis.repositories.enabled=false
#######redis setting#######
spring.redis.host=localhost
spring.redis.port=6379
spring.redis.password=REDISPASSWD_CHANGE
#spring.redis.password=REDISPASSWD_CHANGE
spring.profiles.active=prod
logging.config=classpath:logback-spring.xml
Loading