Skip to content

D
Distributed Computing and Network Fusion System
Commit 9ad2cfc3 authored by wutu's avatar wutu
Browse files
Options

开始完善边缘侧服务功能，完成网络信息查看、集群创建的Controller服务接口。

parent 40e34aff
Show whitespace changes
Inline Side-by-side
Showing with 379 additions and 54 deletions
bishe-edge-center/src/main/java/top/ninwoo/edgecenter/controller/IndexController.java
View file @ 9ad2cfc3
package top.ninwoo.edgecenter.controller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
import top.ninwoo.edgecenter.entity.ClusterConfig;
import top.ninwoo.edgecenter.service.ClusterService;
import top.ninwoo.utils.entity.DockerContainer;
import top.ninwoo.utils.entity.NetworkInfo;
import top.ninwoo.utils.entity.OvsBridge;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
@RestController
public class IndexController {
ConcurrentHashMap<String, NetworkInfo> networkInfoMap = new ConcurrentHashMap<>();
ConcurrentHashMap<String, Future> scheduledFutureMap = new ConcurrentHashMap<>();
ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(8);
@Autowired
ClusterService clusterService;
......@@ -27,4 +34,84 @@ public class IndexController {
List<OvsBridge> bridges = clusterService.getOvsBridges();
return bridges;
}
/**
* 查询容器的网络信息，根据容器的id，获取对应集群的全部节点网络实时情况
* @param clusterId
* @param containerId
* @return
*/
// 这里一次查询需要两秒，需要采用异步进行查询
@RequestMapping("/networkMonitor")
public List<NetworkInfo> getNetworkMonitor(@RequestParam("clusterId") Long clusterId,@RequestParam("containerId") String containerId) {
// 获取全部的容器id
Set<String> cids = clusterService.getContainerIdsByClusterId(clusterId, containerId);
// 根据容器id获取全部的
//List<NetworkInfo> networkTraffic = clusterService.getNetworkTraffic(containerIdsByClusterId);
List<NetworkInfo> networkInfos = new ArrayList<>();
cids.forEach( cid -> {
if(networkInfoMap.containsKey(cid)) {
NetworkInfo networkInfo = networkInfoMap.get(cid);
networkInfos.add(networkInfo);
}
// TODO: 这里还可以再添加新的处理方式
});
return networkInfos;
}
@RequestMapping("/enableNetworkMonitor")
public boolean enableNetworkMonitor(long clusterId, String containerName) {
// 先获取全部的容器id
Set<String> cids = clusterService.getContainerIdsByClusterId(clusterId, containerName);
if(cids == null || cids.isEmpty()) {
return false;
}
cids.forEach(cid -> {
// 为每一个cid容器创建一个定时任务
NetworkMonitorThread networkMonitorThread = new NetworkMonitorThread(cid, networkInfoMap, clusterService);
// 放入线程池执行任务
ScheduledFuture<?> future = scheduledExecutorService.schedule(networkMonitorThread, 1, TimeUnit.SECONDS);
// 保存任务，以方便进行定时任务的关闭
scheduledFutureMap.put(cid, future);
});
return true;
}
@RequestMapping(value = "/createCluster", method = RequestMethod.POST)
public long createCluster(@RequestBody ClusterConfig clusterConfig) {
// 创建一个解析器
long l = clusterService.initContainers(clusterConfig);
// 使用json进行解析
return l;
}
@RequestMapping(value = "/getContainerIds")
public Set<String> getContainerIds(long clusterId, String containerName) {
return clusterService.getContainerIdsByClusterId(clusterId, containerName);
}
private static class NetworkMonitorThread implements Runnable {
private String containerId;
private ConcurrentHashMap<String, NetworkInfo> map;
private ClusterService clusterService;
public NetworkMonitorThread(String containerId, ConcurrentHashMap<String, NetworkInfo> map, ClusterService clusterService) {
this.containerId = containerId;
this.map = map;
this.clusterService = clusterService;
}
@Override
public void run() {
NetworkInfo networkInfo = clusterService.getNetworkInfoByContainerId(containerId);
map.put(containerId, networkInfo);
}
public void setContainerId(String containerId) {
this.containerId = containerId;
}
}
}
bishe-edge-center/src/main/java/top/ninwoo/edgecenter/service/ClusterService.java
View file @ 9ad2cfc3
......@@ -2,6 +2,7 @@ package top.ninwoo.edgecenter.service;
import top.ninwoo.edgecenter.entity.ClusterConfig;
import top.ninwoo.utils.entity.DockerContainer;
import top.ninwoo.utils.entity.NetworkInfo;
import top.ninwoo.utils.entity.OvsBridge;
import java.util.List;
......@@ -30,4 +31,8 @@ public interface ClusterService {
// 获取网桥信息
List<OvsBridge> getOvsBridges();
List<NetworkInfo> getNetworkTraffic(Set<String> containerIds);
NetworkInfo getNetworkInfoByContainerId(String containerId);
}
bishe-edge-center/src/main/java/top/ninwoo/edgecenter/service/impl/ClusterServiceImpl.java
View file @ 9ad2cfc3
......@@ -8,9 +8,11 @@ import top.ninwoo.edgecenter.entity.ClusterConfig;
import top.ninwoo.edgecenter.entity.ContainerDescription;
import top.ninwoo.edgecenter.service.ClusterService;
import top.ninwoo.utils.entity.DockerContainer;
import top.ninwoo.utils.entity.NetworkInfo;
import top.ninwoo.utils.entity.OvsBridge;
import top.ninwoo.utils.service.DockerService;
import top.ninwoo.utils.service.OVSService;
import top.ninwoo.utils.service.TcService;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
......@@ -30,6 +32,10 @@ public class ClusterServiceImpl implements ClusterService {
@Autowired
OVSService ovsService;
// TODO: 这个服务接口可以拆分成两个
@Autowired
TcService tcService;
/**
* 定时更新容器列表
*/
......@@ -176,4 +182,34 @@ public class ClusterServiceImpl implements ClusterService {
LOG.debug("查询ovs网桥");
return ovsService.getOvsBridges();
}
/**
* 根据容器id，获取容器的网络信息
* @param containerIds
* @return
*/
@Override
public List<NetworkInfo> getNetworkTraffic(Set<String> containerIds) {
List<NetworkInfo> networkInfos = new ArrayList<>();
containerIds.forEach(cid -> {
try {
NetworkInfo networkInfo = tcService.networkUsage(cid);
networkInfo.setContainerId(cid);
networkInfos.add(networkInfo);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
return networkInfos;
}
@Override
public NetworkInfo getNetworkInfoByContainerId(String containerId) {
try {
return tcService.networkUsage(containerId);
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
}
\ No newline at end of file
bishe-edge-center/src/test/java/top/ninwoo/edgecenter/ClusterServiceTest.java
View file @ 9ad2cfc3
package top.ninwoo.edgecenter;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
......@@ -64,7 +65,7 @@ public class ClusterServiceTest {
System.out.println(clusterService.getContainerIdsByClusterId(clusterId, "APP"));
}
@Test
@After
public void testRemoveContainers() {
clusterService.removeContainersByClusterId(clusterId);
}
......
bishe-edge-center/src/test/java/top/ninwoo/edgecenter/InDockerTests.java
View file @ 9ad2cfc3
......@@ -153,6 +153,7 @@ public class InDockerTests {
public void testTcLoss() {
// 查看未加限制的网络情况
dockerService.execInDocker(dockerContainer1.getId(), "tc qdisc add dev eth1 root netem loss 10%");
//tcService.addDelay(dockerContainer1.getId());
LOG.info("10%丢包率：" + testLoss());
}
......@@ -181,7 +182,7 @@ public class InDockerTests {
public String pingTest() {
return dockerService.execInDocker(dockerContainer1.getId(), "ping 10.1.100.3 -c 100");
return dockerService.execInDocker(dockerContainer1.getId(), "ping 10.1.100.3 -c 10");
}
@Test
......@@ -255,23 +256,35 @@ public class InDockerTests {
* 这里主要测试接口限速，以及统计实时流量，需要外部辅助执行其他命令
*/
@Test
public void testTCService() {
public void testTCService() throws InterruptedException {
// 限速
System.out.println(tcService.addQos(dockerContainer2.getId(), "4mbit", "400ms"));
//System.out.println(tcService.addQos(dockerContainer2.getId(), "4mbit", "400ms"));
// 测试丢包率
System.out.println(tcService.addDelay(dockerContainer2.getId(), 10));
//dockerService.execInDocker(dockerContainer2.getId(), "tc qdisc add dev eth1 root tbf rate 1mbit limit 10k burst 10k mtu 5000");
//dockerService.execInDocker(dockerContainer1.getId(), "tc qdisc add dev eth1 root handle 1: htb default 12");
//dockerService.execInDocker(dockerContainer1.getId(), "tc class add dev eth1 parent 1: classid 1:12 htb rate 1kbit");
new Thread(new Runnable() {
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 50; i++) {
System.out.println(tcService.networkUsage(dockerContainer2.getId()));
while(!Thread.currentThread().isInterrupted()) {
try {
System.out.println("Network Usage:" + tcService.networkUsage(dockerContainer2.getId()));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}).start();
});
pingTest();
thread.start();
Thread.sleep(1000);
System.out.println(pingTest());
String result = dockerService.execInDocker(dockerContainer2.getId(), "iperf -c 10.1.100.2 -i -t 10".split(" "));
thread.interrupt();
System.out.println(result);
}
......
bishe-utils/src/main/java/top/ninwoo/utils/entity/LinkInfo.java 0 → 100644
View file @ 9ad2cfc3
package top.ninwoo.utils.entity;
/**
* @Author joliu
* @Description
* @Date　Create in 上午10:25 2019/11/5
*/
public class LinkInfo {
private int delay;
private int randomDelay;
private int correlation;
private String policy;
private int loss;
public int getDelay() {
return delay;
}
public void setDelay(int delay) {
this.delay = delay;
}
public int getRandomDelay() {
return randomDelay;
}
public void setRandomDelay(int randomDelay) {
this.randomDelay = randomDelay;
}
public int getCorrelation() {
return correlation;
}
public void setCorrelation(int correlation) {
this.correlation = correlation;
}
public String getPolicy() {
return policy;
}
public void setPolicy(String policy) {
this.policy = policy;
}
public int getLoss() {
return loss;
}
public void setLoss(int loss) {
this.loss = loss;
}
}
bishe-utils/src/main/java/top/ninwoo/utils/entity/NetworkInfo.java 0 → 100644
View file @ 9ad2cfc3
package top.ninwoo.utils.entity;
/**
* @Author joliu
* @Description
* @Date　Create in 下午9:51 2019/11/4
*/
public class NetworkInfo {
private String containerId;
private float curRate;
private float errorRate;
private float dropRate;
public String getContainerId() {
return containerId;
}
public void setContainerId(String containerId) {
this.containerId = containerId;
}
public float getCurRate() {
return curRate;
}
public void setCurRate(float curRate) {
this.curRate = curRate;
}
public float getErrorRate() {
return errorRate;
}
public void setErrorRate(float errorRate) {
this.errorRate = errorRate;
}
public float getDropRate() {
return dropRate;
}
public void setDropRate(float dropRate) {
this.dropRate = dropRate;
}
@Override
public String toString() {
return "NetworkInfo{" +
"curRate=" + curRate +
"%, errorRate=" + errorRate +
"%, dropRate=" + dropRate +
"%}";
}
}
bishe-utils/src/main/java/top/ninwoo/utils/service/TcService.java
View file @ 9ad2cfc3
package top.ninwoo.utils.service;
import top.ninwoo.utils.entity.NetworkInfo;
/**
* @Author joliu
* @Description
......@@ -12,5 +14,11 @@ public interface TcService {
boolean delQos(String containerId);
float networkUsage(String containerId);
NetworkInfo networkUsage(String containerId) throws InterruptedException;
boolean addDelay(String containerId, int delay, int randomDelay, int correlation);
boolean addDelay(String containerId, int loss);
boolean addDelay(String containerId);
}
bishe-utils/src/main/java/top/ninwoo/utils/service/impl/TcServiceImpl.java
View file @ 9ad2cfc3
......@@ -2,11 +2,12 @@ package top.ninwoo.utils.service.impl;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import top.ninwoo.utils.entity.LinkInfo;
import top.ninwoo.utils.entity.NetworkInfo;
import top.ninwoo.utils.service.TcService;
import top.ninwoo.utils.util.DockerUtils;
import top.ninwoo.utils.util.TcUtils;
import java.io.*;
import java.util.concurrent.ConcurrentHashMap;
/**
......@@ -18,6 +19,7 @@ import java.util.concurrent.ConcurrentHashMap;
public class TcServiceImpl implements TcService {
// 这里需要设计成有状态的
private ConcurrentHashMap<String, Integer> qosMap = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, LinkInfo> delayMap = new ConcurrentHashMap<>();
@Autowired
TcUtils tcUtils;
......@@ -28,80 +30,132 @@ public class TcServiceImpl implements TcService {
@Override
public boolean addQos(String containerId, String maxRate, String latency) {
// 添加到map中
if(qosMap.containsKey(containerId)) {
delQos(containerId);
}
qosMap.put(containerId, Integer.parseInt(maxRate.substring(0, maxRate.length() - 4)));
return tcUtils.addQos(containerId, maxRate, latency);
}
@Override
public boolean delQos(String containerId) {
if(qosMap.containsKey(containerId)) {
qosMap.remove(containerId);
return tcUtils.delQos(containerId);
}
return true;
}
/**
* 测试docker容器的网络实时速率
* @param containerId
* @return
* @throws InterruptedException
*/
@Override
public float networkUsage(String containerId) {
// TODO: TOTALBAND需要根据具体情况进行设置
System.out.println("开始收集带宽率");
public NetworkInfo networkUsage(String containerId) throws InterruptedException {
NetworkInfo networkInfo = new NetworkInfo();
float netUsage = 0.0f;
String command = "cat /proc/net/dev";
//第一次采集流量数据
long startTime = System.currentTimeMillis();
String[] lines = dockerUtils.execInDocker(containerId, command.split(" ")).split("\n");
int i = 0;
long inSize1 = 0, outSize1 = 0;
long packetIn1 = 0, packetOut1 = 0;
long errorIn1 = 0, errorOut1 = 0;
long dropIn1 = 0, dropOut1 = 0;
while(i < lines.length){
String line = lines[i++];
line = line.trim();
if(line.startsWith("eth1")){
System.out.println(line);
String[] temp = line.split("\\s+");
//System.out.println("temp: "+temp.length+"temp[0]="+temp[0]);
//这里可能因为不同操作系统 展示的结果不同，导致下标不同，
//自己对照 cat /proc/net/dev 该指令执行后展示出的结构 找到Receive bytes 是数组里第几个元素，替换下标即可
inSize1 = Long.parseLong(temp[1]); //Receive bytes,单位为Byte
outSize1 = Long.parseLong(temp[9]); //Transmit bytes,单位为Byte
packetIn1 = Long.parseLong(temp[2]);
packetOut1 = Long.parseLong(temp[10]);
errorIn1 = Long.parseLong(temp[3]);
errorOut1 = Long.parseLong(temp[11]);
dropIn1 = Long.parseLong(temp[4]);
dropOut1 = Long.parseLong(temp[12]);
break;
}
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
StringWriter sw = new StringWriter();
e.printStackTrace(new PrintWriter(sw));
System.out.println("NetUsage休眠时发生InterruptedException. " + e.getMessage());
System.out.println(sw.toString());
}
//第二次采集流量数据
long endTime = System.currentTimeMillis();
lines = dockerUtils.execInDocker(containerId, command.split(" ")).split("\n");
i = 0;
long inSize2 = 0 ,outSize2 = 0;
long packetIn2 = 0, packetOut2 = 0;
long errorIn2 = 0, errorOut2 = 0;
long dropIn2 = 0, dropOut2 = 0;
while(i < lines.length){
String line = lines[i++];
line = line.trim();
if(line.startsWith("eth1")){
//System.out.println(line);
String[] temp = line.split("\\s+");
//这里数组下标也需要修改
inSize2 = Long.parseLong(temp[1]);
outSize2 = Long.parseLong(temp[9]);
packetIn2 = Long.parseLong(temp[2]);
packetOut2 = Long.parseLong(temp[10]);
errorIn2 = Long.parseLong(temp[3]);
errorOut2 = Long.parseLong(temp[11]);
dropIn2 = Long.parseLong(temp[4]);
dropOut2 = Long.parseLong(temp[12]);
break;
}
}
if(inSize1 != 0 && outSize1 !=0 && inSize2 != 0 && outSize2 !=0){
float interval = (float)(endTime - startTime)/1000;
//网口传输速度,单位为bps
float curRate = (float)(inSize2 - inSize1 + outSize2 - outSize1)*8/(1000000*interval);
float errorRate = (float)(errorIn2 - errorIn1 + errorOut2 - errorOut1) /
(packetIn2 - packetIn1 + packetOut2 - packetOut1);
float dropRate = (float) (dropIn2 - dropIn1 + dropOut2 - dropOut1) /
(packetIn2 - packetIn1 + packetOut2 - packetOut1);
int maxRate = 1000;
if(qosMap.containsKey(containerId)) {
maxRate = qosMap.get(containerId);
}
netUsage = curRate/maxRate;
//System.out.println("本节点网口速度为: " + curRate + "Mbps");
//System.out.println("本节点网络带宽使用率为: " + netUsage);
if(netUsage > 1.0) {
netUsage = 1.0f;
}
networkInfo.setCurRate(curRate);
networkInfo.setDropRate(dropRate);
networkInfo.setErrorRate(errorRate);
}
return networkInfo;
}
@Override
public boolean addDelay(String containerId, int delay, int randomDelay, int correlation) {
String mode = "add";
if (delayMap.containsKey(containerId)) {
mode = "change";
}
return netUsage;
LinkInfo linkInfo = new LinkInfo();
linkInfo.setCorrelation(0);
linkInfo.setPolicy("");
linkInfo.setRandomDelay(randomDelay);
linkInfo.setDelay(delay);
delayMap.put(containerId, linkInfo);
return tcUtils.addDelay(containerId, mode, delay, randomDelay, correlation);
}
// TODO:这里还可以添加更多实现的接口
@Override
public boolean addDelay(String containerId, int loss) {
String mode = "add";
if(delayMap.containsKey(containerId)) {
mode = "change";
}
LinkInfo linkInfo = new LinkInfo();
linkInfo.setLoss(0);
delayMap.put(containerId, linkInfo);
return tcUtils.addDelay(containerId, mode, 0, 0, 0, loss);
}
@Override
public boolean addDelay(String containerId) {
return false;
}
}
\ No newline at end of file
bishe-utils/src/main/java/top/ninwoo/utils/util/TcUtils.java
View file @ 9ad2cfc3
......@@ -11,11 +11,13 @@ public interface TcUtils {
boolean delQos(String containerId);
boolean addDelay(String containerId, int delay);
boolean addDelay(String containerId, String mode, int delay);
boolean addDelay(String containerId, int delay, int randomDelay);
boolean addDelay(String containerId, String mode, int delay, int randomDelay);
boolean addDelay(String containerId, int delay, int randomDelay, int correlation);
boolean addDelay(String containerId, String mode, int delay, int randomDelay, int correlation);
boolean addDelay(String contianerId, int delay, int randomDelay, String policy);
boolean addDelay(String containerId, String mode, int delay, int randomDelay, int correlation, int loss);
boolean addDelay(String contianerId, String mode, int delay, int randomDelay, String policy);
}
bishe-utils/src/main/java/top/ninwoo/utils/util/impl/TcUtilsImpl.java
View file @ 9ad2cfc3
......@@ -61,13 +61,13 @@ public class TcUtilsImpl implements TcUtils {
}
@Override
public boolean addDelay(String containerId, int delay) {
return addDelay(containerId, delay, 0, 0);
public boolean addDelay(String containerId, String mode, int delay) {
return addDelay(containerId, mode, delay, 0, 0);
}
@Override
public boolean addDelay(String containerId, int delay, int randomDelay) {
return addDelay(containerId, delay, randomDelay, 0);
public boolean addDelay(String containerId, String mode, int delay, int randomDelay) {
return addDelay(containerId, mode, delay, randomDelay, 0);
}
/**
......@@ -79,13 +79,25 @@ public class TcUtilsImpl implements TcUtils {
* @return
*/
@Override
public boolean addDelay(String containerId, int delay, int randomDelay, int correlation) {
String cmd = "tc qdisc add dev eth1 root netem delay " + delay + "ms ";
public boolean addDelay(String containerId, String mode, int delay, int randomDelay, int correlation) {
return addDelay(containerId, mode, delay, randomDelay, correlation, 0);
}
@Override
public boolean addDelay(String containerId, String mode, int delay, int randomDelay, int correlation, int loss) {
// tc qdisc add dev eth1 root netem loss 10%
String cmd = "tc qdisc " + mode + " dev eth1 root netem ";
if(delay > 0) {
cmd += delay + "ms ";
}
if(randomDelay > 0) {
cmd += randomDelay + "ms ";
}
if(correlation >0 && correlation < 100) {
cmd += correlation + "%";
cmd += correlation + "% ";
}
if(loss > 0) {
cmd += "loss " + loss + "%";
}
String res = dockerUtils.execInDocker(containerId, cmd);
if("".equals(res)) {
......@@ -93,7 +105,6 @@ public class TcUtilsImpl implements TcUtils {
}
return false;
}
/**
* 正太分布的随机延时
* @param contianerId
......@@ -103,13 +114,13 @@ public class TcUtilsImpl implements TcUtils {
* @return
*/
@Override
public boolean addDelay(String contianerId, int delay, int randomDelay, String policy) {
public boolean addDelay(String contianerId, String mode, int delay, int randomDelay, String policy) {
if(!supportPolicies.contains(policy)) {
return false;
}
String cmd = "tc qdisc change dev eth1 netem delay " + delay + "ms " + randomDelay + "ms " + policy + " normal";
String cmd = "tc qdisc " + mode + " dev eth1 netem delay " + delay + "ms " + randomDelay + "ms " + policy + " normal";
String res = dockerUtils.execInDocker(contianerId, cmd);
if("".equals(res)) {
return true;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment