mirror of
https://github.com/weaveworks/scope.git
synced 2026-07-28 01:31:17 +00:00
plugins: remove plugins
They now are in the weaveworks-plugins organization. https://github.com/weaveworks-plugins - http-request has been updated and renamed to http-statistics: https://github.com/weaveworks-plugins/scope-http-statistics - iowait: https://github.com/weaveworks-plugins/scope-iowait - traffic-control: https://github.com/weaveworks-plugins/scope-traffic-control
This commit is contained in:
Alessandro Puccetti
@@ -1,12 +0,0 @@
|
||||
FROM ubuntu:wily
|
||||
MAINTAINER Weaveworks Inc <help@weave.works>
|
||||
LABEL works.weave.role=system
|
||||
|
||||
# Install BCC
|
||||
RUN apt-key adv --keyserver keyserver.ubuntu.com --recv-keys D4284CDD
|
||||
RUN echo "deb http://52.8.15.63/apt trusty main" | tee /etc/apt/sources.list.d/iovisor.list
|
||||
RUN apt-get update && apt-get install -y libbcc libbcc-examples python-bcc
|
||||
|
||||
# Add our plugin
|
||||
ADD ./http-requests.c ./http-requests.py /usr/bin/
|
||||
ENTRYPOINT ["/usr/bin/http-requests.py"]
|
||||
@@ -1,22 +0,0 @@
|
||||
.PHONY: run clean
|
||||
|
||||
IMAGE=weavescope-http-requests-plugin
|
||||
UPTODATE=.uptodate
|
||||
|
||||
run: $(UPTODATE)
|
||||
docker run --rm -it \
|
||||
--privileged --net=host \
|
||||
-v /lib/modules:/lib/modules \
|
||||
-v /usr/src:/usr/src \
|
||||
-v /sys/kernel/debug/:/sys/kernel/debug/ \
|
||||
-v /var/run/scope/plugins:/var/run/scope/plugins \
|
||||
--name $(IMAGE) \
|
||||
$(IMAGE)
|
||||
|
||||
$(UPTODATE): Dockerfile http-requests.py http-requests.c
|
||||
docker build -t $(IMAGE) .
|
||||
touch $@
|
||||
|
||||
clean:
|
||||
- rm -rf $(UPTODATE)
|
||||
- docker rmi $(IMAGE)
|
||||
@@ -1,162 +0,0 @@
|
||||
#include <linux/skbuff.h>
|
||||
#include <net/sock.h>
|
||||
|
||||
/* Table from (Task group id|Task id) to (Number of received http requests).
|
||||
We need to gather requests per task and not only per task group (i.e. userspace pid)
|
||||
so that entries can be cleared up independently when a task exists.
|
||||
This implies that userspace needs to do the per-process aggregation.
|
||||
*/
|
||||
BPF_HASH(received_http_requests, u64, u64);
|
||||
|
||||
|
||||
/* skb_copy_datagram_iter() (Kernels >= 3.19) is in charge of copying socket
|
||||
buffers from kernel to userspace.
|
||||
|
||||
skb_copy_datagram_iter() has an associated tracepoint
|
||||
(trace_skb_copy_datagram_iovec), which would be more stable than a kprobe but
|
||||
it lacks the offset argument.
|
||||
*/
|
||||
int kprobe__skb_copy_datagram_iter(struct pt_regs *ctx, const struct sk_buff *skb, int offset, void *unused_iovec, int len)
|
||||
{
|
||||
|
||||
/* Inspect the beginning of socket buffers copied to user-space to determine
|
||||
if they correspond to http requests.
|
||||
|
||||
Caveats:
|
||||
|
||||
Requests may not appear at the beginning of a packet due to:
|
||||
* Persistent connections.
|
||||
* Packet fragmentation.
|
||||
|
||||
We could inspect the full packet but:
|
||||
* It's very inefficient.
|
||||
* Examining the non-linear (paginated) area of a socket buffer would be
|
||||
really tricky from ebpf.
|
||||
*/
|
||||
|
||||
/* Verify it's a TCP socket
|
||||
TODO: is it worth caching it in a socket table?
|
||||
*/
|
||||
struct sock *sk = skb->sk;
|
||||
unsigned short skc_family = sk->__sk_common.skc_family;
|
||||
switch (skc_family) {
|
||||
case PF_INET:
|
||||
case PF_INET6:
|
||||
case PF_UNIX:
|
||||
break;
|
||||
default:
|
||||
return 0;
|
||||
}
|
||||
/* The socket type and protocol are not directly addressable since they are
|
||||
bitfields. We access them by assuming sk_write_queue is immediately before
|
||||
them (admittedly pretty hacky).
|
||||
*/
|
||||
unsigned int flags = 0;
|
||||
size_t flags_offset = offsetof(typeof(struct sock), sk_write_queue) + sizeof(sk->sk_write_queue);
|
||||
bpf_probe_read(&flags, sizeof(flags), ((u8*)sk) + flags_offset);
|
||||
u16 sk_type = flags >> 16;
|
||||
if (sk_type != SOCK_STREAM) {
|
||||
return 0;
|
||||
}
|
||||
u8 sk_protocol = flags >> 8 & 0xFF;
|
||||
/* The protocol is unset (IPPROTO_IP) in Unix sockets */
|
||||
if ( (sk_protocol != IPPROTO_TCP) && ((skc_family == PF_UNIX) && (sk_protocol != IPPROTO_IP)) ) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Inline implementation of skb_headlen() */
|
||||
unsigned int head_len = skb->len - skb->data_len;
|
||||
/* http://stackoverflow.com/questions/25047905/http-request-minimum-size-in-bytes
|
||||
minimum length of http request is always greater than 7 bytes
|
||||
*/
|
||||
unsigned int available_data = head_len - offset;
|
||||
if (available_data < 7) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Check if buffer begins with a method name followed by a space.
|
||||
|
||||
To avoid false positives it would be good to do a deeper inspection
|
||||
(i.e. fully ensure a 'Method SP Request-URI SP HTTP-Version CRLF'
|
||||
structure) but loops are not allowed in ebpf, making variable-size-data
|
||||
parsers infeasible.
|
||||
*/
|
||||
u8 data[8] = {};
|
||||
if (available_data >= 8) {
|
||||
/* We have confirmed having access to 7 bytes, but need 8 bytes to check the
|
||||
space after OPTIONS. bpf_probe_read() requires its second argument to be
|
||||
an immediate, so we obtain the data in this unsexy way.
|
||||
*/
|
||||
bpf_probe_read(&data, 8, skb->data + offset);
|
||||
} else {
|
||||
bpf_probe_read(&data, 7, skb->data + offset);
|
||||
}
|
||||
|
||||
switch (data[0]) {
|
||||
/* DELETE */
|
||||
case 'D':
|
||||
if ((data[1] != 'E') || (data[2] != 'L') || (data[3] != 'E') || (data[4] != 'T') || (data[5] != 'E') || (data[6] != ' ')) {
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
|
||||
/* GET */
|
||||
case 'G':
|
||||
if ((data[1] != 'E') || (data[2] != 'T') || (data[3] != ' ')) {
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
|
||||
/* HEAD */
|
||||
case 'H':
|
||||
if ((data[1] != 'E') || (data[2] != 'A') || (data[3] != 'D') || (data[4] != ' ')) {
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
|
||||
/* OPTIONS */
|
||||
case 'O':
|
||||
if (available_data < 8 || (data[1] != 'P') || (data[2] != 'T') || (data[3] != 'I') || (data[4] != 'O') || (data[5] != 'N') || (data[6] != 'S') || (data[7] != ' ')) {
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
|
||||
/* PATCH/POST/PUT */
|
||||
case 'P':
|
||||
switch (data[1]) {
|
||||
case 'A':
|
||||
if ((data[2] != 'T') || (data[3] != 'C') || (data[4] != 'H') || (data[5] != ' ')) {
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
case 'O':
|
||||
if ((data[2] != 'S') || (data[3] != 'T') || (data[4] != ' ')) {
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
case 'U':
|
||||
if ((data[2] != 'T') || (data[3] != ' ')) {
|
||||
return 0;
|
||||
}
|
||||
break;
|
||||
}
|
||||
break;
|
||||
|
||||
default:
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* Finally, bump the request counter for current task */
|
||||
u64 pid_tgid = bpf_get_current_pid_tgid();
|
||||
received_http_requests.increment(pid_tgid);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
/* Clear out request count entries of tasks on exit */
|
||||
int kprobe__do_exit(struct pt_regs *ctx) {
|
||||
u64 pid_tgid = bpf_get_current_pid_tgid();
|
||||
received_http_requests.delete(&pid_tgid);
|
||||
return 0;
|
||||
}
|
||||
@@ -1,195 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
import bcc
|
||||
|
||||
import time
|
||||
import collections
|
||||
import datetime
|
||||
import os
|
||||
import signal
|
||||
import errno
|
||||
import json
|
||||
import urlparse
|
||||
import threading
|
||||
import socket
|
||||
import BaseHTTPServer
|
||||
import SocketServer
|
||||
import sys
|
||||
|
||||
EBPF_FILE = "http-requests.c"
|
||||
EBPF_TABLE_NAME = "received_http_requests"
|
||||
PLUGIN_ID="http-requests"
|
||||
PLUGIN_UNIX_SOCK = "/var/run/scope/plugins/" + PLUGIN_ID + ".sock"
|
||||
|
||||
class KernelInspector(threading.Thread):
|
||||
def __init__(self):
|
||||
super(KernelInspector, self).__init__()
|
||||
self.bpf = bcc.BPF(EBPF_FILE)
|
||||
self.http_rate_per_pid = dict()
|
||||
self.lock = threading.Lock()
|
||||
|
||||
def update_http_rate_per_pid(self, last_req_count_snapshot):
|
||||
# Aggregate the kernel's per-task http request counts into userland's
|
||||
# per-process counts
|
||||
req_count_table = self.bpf.get_table(EBPF_TABLE_NAME)
|
||||
new_req_count_snapshot = collections.defaultdict(int)
|
||||
for pid_tgid, req_count in req_count_table.iteritems():
|
||||
# Note that the kernel's tgid maps into userland's pid
|
||||
# (not to be confused by the kernel's pid, which is
|
||||
# the unique identifier of a kernel task)
|
||||
pid = pid_tgid.value >> 32
|
||||
new_req_count_snapshot[pid] += req_count.value
|
||||
|
||||
# Compute request rate
|
||||
new_http_rate_per_pid = dict()
|
||||
for pid, req_count in new_req_count_snapshot.iteritems():
|
||||
request_delta = req_count
|
||||
if pid in last_req_count_snapshot:
|
||||
request_delta -= last_req_count_snapshot[pid]
|
||||
new_http_rate_per_pid[pid] = request_delta
|
||||
|
||||
self.lock.acquire()
|
||||
self.http_rate_per_pid = new_http_rate_per_pid
|
||||
self.lock.release()
|
||||
|
||||
return new_req_count_snapshot
|
||||
|
||||
def on_http_rate_per_pid(self, f):
|
||||
self.lock.acquire()
|
||||
r = f(self.http_rate_per_pid)
|
||||
self.lock.release()
|
||||
return r
|
||||
|
||||
def run(self):
|
||||
# Compute request rates based on the requests counts from the last
|
||||
# second. It would be simpler to clear the table, wait one second but
|
||||
# clear() is expensive (each entry is individually cleared with a system
|
||||
# call) and less robust (it contends with the increments done by the
|
||||
# kernel probe).
|
||||
req_count_snapshot = collections.defaultdict(int)
|
||||
while True:
|
||||
time.sleep(1)
|
||||
req_count_snapshot = self.update_http_rate_per_pid(req_count_snapshot)
|
||||
|
||||
|
||||
class PluginRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
|
||||
protocol_version = 'HTTP/1.1'
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.request_log = ''
|
||||
BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args, **kwargs)
|
||||
|
||||
def do_GET(self):
|
||||
self.log_extra = ''
|
||||
path = urlparse.urlparse(self.path)[2].lower()
|
||||
if path == '/report':
|
||||
self.do_report()
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.send_header('Content-length', 0)
|
||||
self.end_headers()
|
||||
|
||||
def get_process_nodes(self, http_rate_per_pid):
|
||||
# Get current timestamp in RFC3339
|
||||
date = datetime.datetime.utcnow()
|
||||
date = date.isoformat('T') + 'Z'
|
||||
process_nodes = dict()
|
||||
for pid, http_rate in http_rate_per_pid.iteritems():
|
||||
# print "\t%-10s %s" % (pid , http_rate)
|
||||
node_key = "%s;%d" % (self.server.hostname, pid)
|
||||
process_nodes[node_key] = {
|
||||
'metrics': {
|
||||
'http_requests_per_second': {
|
||||
'samples': [{
|
||||
'date': date,
|
||||
'value': float(http_rate),
|
||||
}]
|
||||
}
|
||||
}
|
||||
}
|
||||
return process_nodes
|
||||
|
||||
def do_report(self):
|
||||
kernel_inspector = self.server.kernel_inspector
|
||||
process_nodes = kernel_inspector.on_http_rate_per_pid(self.get_process_nodes)
|
||||
report = {
|
||||
'Process': {
|
||||
'nodes': process_nodes,
|
||||
'metric_templates': {
|
||||
'http_requests_per_second': {
|
||||
'id': 'http_requests_per_second',
|
||||
'label': 'HTTP Req/Second',
|
||||
'priority': 0.1,
|
||||
}
|
||||
}
|
||||
},
|
||||
'Plugins': [
|
||||
{
|
||||
'id': PLUGIN_ID,
|
||||
'label': 'HTTP Requests',
|
||||
'description': 'Adds http request metrics to processes',
|
||||
'interfaces': ['reporter'],
|
||||
'api_version': '1',
|
||||
}
|
||||
]
|
||||
}
|
||||
body = json.dumps(report)
|
||||
self.request_log = "resp_size=%d, resp_entry_count=%d" % (len(body), len(process_nodes))
|
||||
self.send_response(200)
|
||||
self.send_header('Content-type', 'application/json')
|
||||
self.send_header('Content-length', len(body))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def log_request(self, code='-', size='-'):
|
||||
request_log = ''
|
||||
if self.request_log:
|
||||
request_log = ' (%s)' % self.request_log
|
||||
self.log_message('"%s" %s %s%s',
|
||||
self.requestline, str(code), str(size), request_log)
|
||||
|
||||
|
||||
class PluginServer(SocketServer.ThreadingUnixStreamServer):
|
||||
daemon_threads = True
|
||||
|
||||
def __init__(self, socket_file, kernel_inspector):
|
||||
mkdir_p(os.path.dirname(socket_file))
|
||||
self.socket_file = socket_file
|
||||
self.delete_socket_file()
|
||||
self.kernel_inspector = kernel_inspector
|
||||
self.hostname = socket.gethostname()
|
||||
SocketServer.UnixStreamServer.__init__(self, socket_file, PluginRequestHandler)
|
||||
|
||||
def finish_request(self, request, _):
|
||||
# Make the logger happy by providing a phony client_address
|
||||
self.RequestHandlerClass(request, '-', self)
|
||||
|
||||
def delete_socket_file(self):
|
||||
if os.path.exists(self.socket_file):
|
||||
os.remove(self.socket_file)
|
||||
|
||||
|
||||
def mkdir_p(path):
|
||||
try:
|
||||
os.makedirs(path)
|
||||
except OSError as exc:
|
||||
if exc.errno == errno.EEXIST and os.path.isdir(path):
|
||||
pass
|
||||
else:
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
kernel_inspector = KernelInspector()
|
||||
kernel_inspector.setDaemon(True)
|
||||
kernel_inspector.start()
|
||||
plugin_server = PluginServer(PLUGIN_UNIX_SOCK, kernel_inspector)
|
||||
def sig_handler(b, a):
|
||||
plugin_server.delete_socket_file()
|
||||
exit(0)
|
||||
signal.signal(signal.SIGTERM, sig_handler)
|
||||
signal.signal(signal.SIGINT, sig_handler)
|
||||
try:
|
||||
plugin_server.serve_forever()
|
||||
except:
|
||||
plugin_server.delete_socket_file()
|
||||
raise
|
||||
1
examples/plugins/iowait/.gitignore
vendored
1
examples/plugins/iowait/.gitignore
vendored
@@ -1 +0,0 @@
|
||||
iowait
|
||||
@@ -1,6 +0,0 @@
|
||||
FROM alpine:3.3
|
||||
MAINTAINER Weaveworks Inc <help@weave.works>
|
||||
LABEL works.weave.role=system
|
||||
COPY ./iowait /usr/bin/iowait
|
||||
RUN mkdir /lib64 && ln -s /lib/libc.musl-x86_64.so.1 /lib64/ld-linux-x86-64.so.2
|
||||
ENTRYPOINT ["/usr/bin/iowait"]
|
||||
@@ -1,25 +0,0 @@
|
||||
.PHONY: run clean
|
||||
|
||||
SUDO=$(shell docker info >/dev/null 2>&1 || echo "sudo -E")
|
||||
EXE=iowait
|
||||
IMAGE=weavescope-iowait-plugin
|
||||
UPTODATE=.$(EXE).uptodate
|
||||
|
||||
run: $(UPTODATE)
|
||||
# --net=host gives us the remote hostname, in case we're being launched against a non-local docker host.
|
||||
# We could also pass in the `-hostname=foo` flag, but that doesn't work against a remote docker host.
|
||||
$(SUDO) docker run --rm -it \
|
||||
--net=host \
|
||||
-v /var/run/scope/plugins:/var/run/scope/plugins \
|
||||
--name $(IMAGE) $(IMAGE)
|
||||
|
||||
$(UPTODATE): $(EXE) Dockerfile
|
||||
$(SUDO) docker build -t $(IMAGE) .
|
||||
touch $@
|
||||
|
||||
$(EXE): main.go
|
||||
$(SUDO) docker run --rm -v "$$PWD":/usr/src/$(EXE) -w /usr/src/$(EXE) golang:1.6 go build -v
|
||||
|
||||
clean:
|
||||
- rm -rf $(UPTODATE) $(EXE)
|
||||
- $(SUDO) docker rmi $(IMAGE)
|
||||
@@ -1,384 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
func main() {
|
||||
hostname, _ := os.Hostname()
|
||||
var (
|
||||
addr = flag.String("addr", "/var/run/scope/plugins/iowait.sock", "unix socket to listen for connections on")
|
||||
hostID = flag.String("hostname", hostname, "hostname of the host running this plugin")
|
||||
)
|
||||
flag.Parse()
|
||||
|
||||
log.Printf("Starting on %s...\n", *hostID)
|
||||
|
||||
// Check we can get the iowait for the system
|
||||
_, err := iowait()
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
os.Remove(*addr)
|
||||
interrupt := make(chan os.Signal, 1)
|
||||
signal.Notify(interrupt, os.Interrupt)
|
||||
go func() {
|
||||
<-interrupt
|
||||
os.Remove(*addr)
|
||||
os.Exit(0)
|
||||
}()
|
||||
|
||||
listener, err := net.Listen("unix", *addr)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
listener.Close()
|
||||
os.Remove(*addr)
|
||||
}()
|
||||
|
||||
log.Printf("Listening on: unix://%s", *addr)
|
||||
|
||||
plugin := &Plugin{HostID: *hostID}
|
||||
http.HandleFunc("/report", plugin.Report)
|
||||
http.HandleFunc("/control", plugin.Control)
|
||||
if err := http.Serve(listener, nil); err != nil {
|
||||
log.Printf("error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Plugin groups the methods a plugin needs
|
||||
type Plugin struct {
|
||||
HostID string
|
||||
|
||||
lock sync.Mutex
|
||||
iowaitMode bool
|
||||
}
|
||||
|
||||
type request struct {
|
||||
NodeID string
|
||||
Control string
|
||||
}
|
||||
|
||||
type response struct {
|
||||
ShortcutReport *report `json:"shortcutReport,omitempty"`
|
||||
}
|
||||
|
||||
type report struct {
|
||||
Host topology
|
||||
Plugins []pluginSpec
|
||||
}
|
||||
|
||||
type topology struct {
|
||||
Nodes map[string]node `json:"nodes"`
|
||||
MetricTemplates map[string]metricTemplate `json:"metric_templates"`
|
||||
Controls map[string]control `json:"controls"`
|
||||
}
|
||||
|
||||
type node struct {
|
||||
Metrics map[string]metric `json:"metrics"`
|
||||
LatestControls map[string]controlEntry `json:"latestControls,omitempty"`
|
||||
}
|
||||
|
||||
type metric struct {
|
||||
Samples []sample `json:"samples,omitempty"`
|
||||
Min float64 `json:"min"`
|
||||
Max float64 `json:"max"`
|
||||
}
|
||||
|
||||
type sample struct {
|
||||
Date time.Time `json:"date"`
|
||||
Value float64 `json:"value"`
|
||||
}
|
||||
|
||||
type controlEntry struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Value controlData `json:"value"`
|
||||
}
|
||||
|
||||
type controlData struct {
|
||||
Dead bool `json:"dead"`
|
||||
}
|
||||
|
||||
type metricTemplate struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label,omitempty"`
|
||||
Format string `json:"format,omitempty"`
|
||||
Priority float64 `json:"priority,omitempty"`
|
||||
}
|
||||
|
||||
type control struct {
|
||||
ID string `json:"id"`
|
||||
Human string `json:"human"`
|
||||
Icon string `json:"icon"`
|
||||
Rank int `json:"rank"`
|
||||
}
|
||||
|
||||
type pluginSpec struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Interfaces []string `json:"interfaces"`
|
||||
APIVersion string `json:"api_version,omitempty"`
|
||||
}
|
||||
|
||||
func (p *Plugin) makeReport() (*report, error) {
|
||||
metrics, err := p.metrics()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rpt := &report{
|
||||
Host: topology{
|
||||
Nodes: map[string]node{
|
||||
p.getTopologyHost(): {
|
||||
Metrics: metrics,
|
||||
LatestControls: p.latestControls(),
|
||||
},
|
||||
},
|
||||
MetricTemplates: p.metricTemplates(),
|
||||
Controls: p.controls(),
|
||||
},
|
||||
Plugins: []pluginSpec{
|
||||
{
|
||||
ID: "iowait",
|
||||
Label: "iowait",
|
||||
Description: "Adds a graph of CPU IO Wait to hosts",
|
||||
Interfaces: []string{"reporter", "controller"},
|
||||
APIVersion: "1",
|
||||
},
|
||||
},
|
||||
}
|
||||
return rpt, nil
|
||||
}
|
||||
|
||||
func (p *Plugin) metrics() (map[string]metric, error) {
|
||||
value, err := p.metricValue()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
id, _ := p.metricIDAndName()
|
||||
metrics := map[string]metric{
|
||||
id: {
|
||||
Samples: []sample{
|
||||
{
|
||||
Date: time.Now(),
|
||||
Value: value,
|
||||
},
|
||||
},
|
||||
Min: 0,
|
||||
Max: 100,
|
||||
},
|
||||
}
|
||||
return metrics, nil
|
||||
}
|
||||
|
||||
func (p *Plugin) latestControls() map[string]controlEntry {
|
||||
ts := time.Now()
|
||||
ctrls := map[string]controlEntry{}
|
||||
for _, details := range p.allControlDetails() {
|
||||
ctrls[details.id] = controlEntry{
|
||||
Timestamp: ts,
|
||||
Value: controlData{
|
||||
Dead: details.dead,
|
||||
},
|
||||
}
|
||||
}
|
||||
return ctrls
|
||||
}
|
||||
|
||||
func (p *Plugin) metricTemplates() map[string]metricTemplate {
|
||||
id, name := p.metricIDAndName()
|
||||
return map[string]metricTemplate{
|
||||
id: {
|
||||
ID: id,
|
||||
Label: name,
|
||||
Format: "percent",
|
||||
Priority: 0.1,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Plugin) controls() map[string]control {
|
||||
ctrls := map[string]control{}
|
||||
for _, details := range p.allControlDetails() {
|
||||
ctrls[details.id] = control{
|
||||
ID: details.id,
|
||||
Human: details.human,
|
||||
Icon: details.icon,
|
||||
Rank: 1,
|
||||
}
|
||||
}
|
||||
return ctrls
|
||||
}
|
||||
|
||||
// Report is called by scope when a new report is needed. It is part of the
|
||||
// "reporter" interface, which all plugins must implement.
|
||||
func (p *Plugin) Report(w http.ResponseWriter, r *http.Request) {
|
||||
p.lock.Lock()
|
||||
defer p.lock.Unlock()
|
||||
log.Println(r.URL.String())
|
||||
rpt, err := p.makeReport()
|
||||
if err != nil {
|
||||
log.Printf("error: %v", err)
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
raw, err := json.Marshal(*rpt)
|
||||
if err != nil {
|
||||
log.Printf("error: %v", err)
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write(raw)
|
||||
}
|
||||
|
||||
// Control is called by scope when a control is activated. It is part
|
||||
// of the "controller" interface.
|
||||
func (p *Plugin) Control(w http.ResponseWriter, r *http.Request) {
|
||||
p.lock.Lock()
|
||||
defer p.lock.Unlock()
|
||||
log.Println(r.URL.String())
|
||||
xreq := request{}
|
||||
err := json.NewDecoder(r.Body).Decode(&xreq)
|
||||
if err != nil {
|
||||
log.Printf("Bad request: %v", err)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
thisNodeID := p.getTopologyHost()
|
||||
if xreq.NodeID != thisNodeID {
|
||||
log.Printf("Bad nodeID, expected %q, got %q", thisNodeID, xreq.NodeID)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
expectedControlID, _, _ := p.controlDetails()
|
||||
if expectedControlID != xreq.Control {
|
||||
log.Printf("Bad control, expected %q, got %q", expectedControlID, xreq.Control)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
p.iowaitMode = !p.iowaitMode
|
||||
rpt, err := p.makeReport()
|
||||
if err != nil {
|
||||
log.Printf("error: %v", err)
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
res := response{ShortcutReport: rpt}
|
||||
raw, err := json.Marshal(res)
|
||||
if err != nil {
|
||||
log.Printf("error: %v", err)
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write(raw)
|
||||
}
|
||||
|
||||
func (p *Plugin) getTopologyHost() string {
|
||||
return fmt.Sprintf("%s;<host>", p.HostID)
|
||||
}
|
||||
|
||||
func (p *Plugin) metricIDAndName() (string, string) {
|
||||
if p.iowaitMode {
|
||||
return "iowait", "IO Wait"
|
||||
}
|
||||
return "idle", "Idle"
|
||||
}
|
||||
|
||||
func (p *Plugin) metricValue() (float64, error) {
|
||||
if p.iowaitMode {
|
||||
return iowait()
|
||||
}
|
||||
return idle()
|
||||
}
|
||||
|
||||
type controlDetails struct {
|
||||
id string
|
||||
human string
|
||||
icon string
|
||||
dead bool
|
||||
}
|
||||
|
||||
func (p *Plugin) allControlDetails() []controlDetails {
|
||||
return []controlDetails{
|
||||
{
|
||||
id: "switchToIdle",
|
||||
human: "Switch to idle",
|
||||
icon: "fa-beer",
|
||||
dead: !p.iowaitMode,
|
||||
},
|
||||
{
|
||||
id: "switchToIOWait",
|
||||
human: "Switch to IO wait",
|
||||
icon: "fa-hourglass",
|
||||
dead: p.iowaitMode,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Plugin) controlDetails() (string, string, string) {
|
||||
for _, details := range p.allControlDetails() {
|
||||
if !details.dead {
|
||||
return details.id, details.human, details.icon
|
||||
}
|
||||
}
|
||||
return "", "", ""
|
||||
}
|
||||
|
||||
func iowait() (float64, error) {
|
||||
return iostatValue(3)
|
||||
}
|
||||
|
||||
func idle() (float64, error) {
|
||||
return iostatValue(5)
|
||||
}
|
||||
|
||||
func iostatValue(idx int) (float64, error) {
|
||||
values, err := iostat()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if idx >= len(values) {
|
||||
return 0, fmt.Errorf("invalid iostat field index %d", idx)
|
||||
}
|
||||
|
||||
return strconv.ParseFloat(values[idx], 64)
|
||||
}
|
||||
|
||||
// Get the latest iostat values
|
||||
func iostat() ([]string, error) {
|
||||
out, err := exec.Command("iostat", "-c").Output()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("iowait: %v", err)
|
||||
}
|
||||
|
||||
// Linux 4.2.0-25-generic (a109563eab38) 04/01/16 _x86_64_(4 CPU)
|
||||
//
|
||||
// avg-cpu: %user %nice %system %iowait %steal %idle
|
||||
// 2.37 0.00 1.58 0.01 0.00 96.04
|
||||
lines := strings.Split(string(out), "\n")
|
||||
if len(lines) < 4 {
|
||||
return nil, fmt.Errorf("iowait: unexpected output: %q", out)
|
||||
}
|
||||
|
||||
values := strings.Fields(lines[3])
|
||||
if len(values) != 6 {
|
||||
return nil, fmt.Errorf("iowait: unexpected output: %q", out)
|
||||
}
|
||||
return values, nil
|
||||
}
|
||||
1
examples/plugins/traffic-control/.gitignore
vendored
1
examples/plugins/traffic-control/.gitignore
vendored
@@ -1 +0,0 @@
|
||||
/traffic-control
|
||||
@@ -1,7 +0,0 @@
|
||||
FROM alpine:3.3
|
||||
MAINTAINER Weaveworks Inc <help@weave.works>
|
||||
LABEL works.weave.role=system
|
||||
COPY ./traffic-control /usr/bin/traffic-control
|
||||
RUN mkdir /lib64 && ln -s /lib/libc.musl-x86_64.so.1 /lib64/ld-linux-x86-64.so.2
|
||||
RUN apk add --update iproute2 && rm -rf /var/cache/apk/*
|
||||
ENTRYPOINT ["/usr/bin/traffic-control"]
|
||||
@@ -1,33 +0,0 @@
|
||||
.PHONY: run clean
|
||||
|
||||
SUDO=$(shell docker info >/dev/null 2>&1 || echo "sudo -E")
|
||||
EXE=traffic-control
|
||||
IMAGE=weaveworks/scope-$(EXE)-plugin
|
||||
NAME=weaveworks-scope-${EXE}-plugin
|
||||
UPTODATE=.$(EXE).uptodate
|
||||
|
||||
run: $(UPTODATE)
|
||||
# --net=host gives us the remote hostname, in case we're being launched against a non-local docker host.
|
||||
# We could also pass in the `-hostname=foo` flag, but that doesn't work against a remote docker host.
|
||||
$(SUDO) docker run --rm -it \
|
||||
--net=host \
|
||||
--pid=host \
|
||||
--privileged \
|
||||
-v /var/run:/var/run \
|
||||
--name $(NAME) $(IMAGE)
|
||||
|
||||
$(UPTODATE): $(EXE) Dockerfile
|
||||
$(SUDO) docker build -t $(IMAGE) .
|
||||
touch $@
|
||||
|
||||
$(EXE): $(shell find . -name *.go)
|
||||
$(SUDO) docker run --rm \
|
||||
-v "$$PWD":/go/src/hosting/org/$(EXE) \
|
||||
-v $(shell pwd)/../../../vendor:/go/src/hosting/org/$(EXE)/vendor \
|
||||
-w /go/src/hosting/org/$(EXE) \
|
||||
golang:1.6 go build -v
|
||||
|
||||
|
||||
clean:
|
||||
- rm -rf $(UPTODATE) $(EXE)
|
||||
- $(SUDO) docker rmi $(IMAGE)
|
||||
@@ -1,17 +0,0 @@
|
||||
# Traffic Control Plugin
|
||||
|
||||
The Traffic Control plugin allows to modify the performance parameters of container's network interfaces. The following images show a simple example of how **status** and **controls** are displayed in scope UI.
|
||||
|
||||
<img src="imgs/container_view.png" width="200" alt="Scope Probe plugin screenshot" align="center">
|
||||
|
||||
## Visualization
|
||||
|
||||
The parameters are shown in a table named **Traffic Control**. The plugin shows the values of latency and packet loss that are enforced on the network interface. The "-" mean that no value is set for that parameter, latency is displayed in *milliseconds* (ms) and packet loss in *percentage*.
|
||||
|
||||
## Controls
|
||||
|
||||
The Traffic Controls plugin provides a simple interface to change the value of latency (hourglass buttons) and packet loss (scissor button) or remove value that was set (circled cross button). Such buttons are displayed on the top of the container detailed view, just above the STATUS section (See picture below, control are shown inside the red rectangle).
|
||||
|
||||
<img src="imgs/controls.png" width="400" alt="Scope Probe plugin screenshot" align="center">
|
||||
|
||||
The *hourglass* buttons control the latency value, from left to right they set: 2000ms, 1000ms, and 500ms. The *scissor* button controls the packet loss value, it set a 10% packet loss. The *circled cross* button clear any previous settings.
|
||||
@@ -1,144 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
log "github.com/Sirupsen/logrus"
|
||||
docker "github.com/fsouza/go-dockerclient"
|
||||
)
|
||||
|
||||
// DockerClient internal data structure
|
||||
type DockerClient struct {
|
||||
store *Store
|
||||
client *docker.Client
|
||||
}
|
||||
|
||||
// NewDockerClient instantiates a new DockerClient
|
||||
func NewDockerClient(store *Store) (*DockerClient, error) {
|
||||
dc, err := docker.NewClient("unix:///var/run/docker.sock")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to connect to docker daemon: %v", err)
|
||||
}
|
||||
return &DockerClient{
|
||||
store: store,
|
||||
client: dc,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Start docker client
|
||||
func (c *DockerClient) Start() {
|
||||
for {
|
||||
func() {
|
||||
events := make(chan *docker.APIEvents)
|
||||
if err := c.client.AddEventListener(events); err != nil {
|
||||
log.Error(err)
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
if err := c.client.RemoveEventListener(events); err != nil {
|
||||
log.Error(err)
|
||||
}
|
||||
}()
|
||||
if err := c.getContainers(); err != nil {
|
||||
log.Error(err)
|
||||
return
|
||||
}
|
||||
for {
|
||||
event, ok := <-events
|
||||
if !ok {
|
||||
log.Error("event listener unexpectedly disconnected")
|
||||
return
|
||||
}
|
||||
c.handleEvent(event)
|
||||
}
|
||||
}()
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *DockerClient) getContainers() error {
|
||||
apiContainers, err := c.client.ListContainers(docker.ListContainersOptions{All: true})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, apiContainer := range apiContainers {
|
||||
containerState, err := c.getContainerState(apiContainer.ID)
|
||||
if err != nil {
|
||||
log.Error(err)
|
||||
continue
|
||||
}
|
||||
state := Destroyed
|
||||
switch {
|
||||
case containerState.Dead || containerState.Paused || containerState.Restarting || containerState.OOMKilled:
|
||||
state = Stopped
|
||||
case containerState.Running:
|
||||
state = Running
|
||||
}
|
||||
c.updateContainer(apiContainer.ID, state, containerState.Pid)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *DockerClient) handleEvent(event *docker.APIEvents) {
|
||||
var state State
|
||||
switch event.Status {
|
||||
case "create":
|
||||
state = Created
|
||||
case "destroy":
|
||||
state = Destroyed
|
||||
case "start", "unpause":
|
||||
state = Running
|
||||
case "die", "pause":
|
||||
state = Stopped
|
||||
default:
|
||||
return
|
||||
}
|
||||
pid, err := c.getContainerPID(event.ID)
|
||||
if err != nil {
|
||||
log.Error(err)
|
||||
return
|
||||
}
|
||||
c.updateContainer(event.ID, state, pid)
|
||||
}
|
||||
|
||||
func (c *DockerClient) getContainerPID(containerID string) (int, error) {
|
||||
containerState, err := c.getContainerState(containerID)
|
||||
if containerState == nil {
|
||||
return 0, err
|
||||
}
|
||||
return containerState.Pid, nil
|
||||
}
|
||||
|
||||
func (c *DockerClient) getContainerState(containerID string) (*docker.State, error) {
|
||||
dockerContainer, err := c.getContainer(containerID)
|
||||
if dockerContainer == nil {
|
||||
return nil, err
|
||||
}
|
||||
return &dockerContainer.State, nil
|
||||
}
|
||||
|
||||
func (c *DockerClient) getContainer(containerID string) (*docker.Container, error) {
|
||||
dockerContainer, err := c.client.InspectContainer(containerID)
|
||||
if err != nil {
|
||||
if _, ok := err.(*docker.NoSuchContainer); !ok {
|
||||
return nil, err
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
return dockerContainer, nil
|
||||
}
|
||||
|
||||
func (c *DockerClient) updateContainer(containerID string, state State, pid int) {
|
||||
if state == Destroyed {
|
||||
c.store.DeleteContainer(containerID)
|
||||
return
|
||||
}
|
||||
cont := Container{
|
||||
State: state,
|
||||
PID: pid,
|
||||
}
|
||||
c.store.SetContainer(containerID, cont)
|
||||
}
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 145 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 49 KiB |
@@ -1,227 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
|
||||
log "github.com/Sirupsen/logrus"
|
||||
)
|
||||
|
||||
// TODO:
|
||||
//
|
||||
// do not try to install the qdics on the network interface every
|
||||
// time, skip this step if it is already installed (currently we do
|
||||
// "replace" instead of "add", but this check may be a way of avoiding
|
||||
// more one-time installation steps in future).
|
||||
//
|
||||
// somehow inform the user about the current traffic control state
|
||||
// (either add some metadata about latency or maybe add background
|
||||
// color to buttons to denote whether the button is active); this may
|
||||
// involve sending shortcut reports as a part of a response to the
|
||||
// control request
|
||||
//
|
||||
// detect if ip and tc binaries are in $PATH
|
||||
//
|
||||
// detect if required sch_netem kernel module is loaded; note that in
|
||||
// some (rare) cases this might be compiled in the kernel instead of
|
||||
// being a separate module; probably check if tc works, if it does not
|
||||
// return something like "not implemented".
|
||||
//
|
||||
// add traffic control on ingress traffic too (ifb kernel module will
|
||||
// be required)
|
||||
//
|
||||
// currently we can control latency, add controls for packet loss and
|
||||
// bandwidth
|
||||
//
|
||||
// port to eBPF?
|
||||
|
||||
type containerClient interface {
|
||||
Start()
|
||||
}
|
||||
|
||||
// Plugin is the internal data structure
|
||||
type Plugin struct {
|
||||
reporter *Reporter
|
||||
|
||||
clients []containerClient
|
||||
}
|
||||
|
||||
// TrafficControlStatus keeps track of parameters status
|
||||
type TrafficControlStatus struct {
|
||||
latency string
|
||||
packetLoss string
|
||||
}
|
||||
|
||||
// String is useful to easily create a string of the traffic control plugin internal status.
|
||||
// Useful for debugging
|
||||
func (tcs *TrafficControlStatus) String() string {
|
||||
return fmt.Sprintf("%s %s", tcs.latency, tcs.packetLoss)
|
||||
}
|
||||
|
||||
// SetLatency sets the latency value
|
||||
// the convention is that empty latency is represented by '-'
|
||||
func (tcs *TrafficControlStatus) SetLatency(latency string) {
|
||||
if latency == "" {
|
||||
tcs.latency = "-"
|
||||
}
|
||||
tcs.latency = latency
|
||||
}
|
||||
|
||||
// SetPacketLoss sets the packet loss value
|
||||
// the convention is that empty packet loss is represented by '-'
|
||||
func (tcs *TrafficControlStatus) SetPacketLoss(packetLoss string) {
|
||||
if packetLoss == "" {
|
||||
tcs.packetLoss = "-"
|
||||
}
|
||||
tcs.packetLoss = packetLoss
|
||||
}
|
||||
|
||||
// TrafficControlStatusInit initializes with the convention that empty values are '-'
|
||||
func TrafficControlStatusInit() *TrafficControlStatus {
|
||||
return &TrafficControlStatus{
|
||||
latency: "-",
|
||||
packetLoss: "-",
|
||||
}
|
||||
}
|
||||
|
||||
// TrafficControlStatusCache implements status caching
|
||||
var trafficControlStatusCache map[string]*TrafficControlStatus
|
||||
|
||||
func main() {
|
||||
const socket = "/var/run/scope/plugins/traffic-control.sock"
|
||||
|
||||
// Handle the exit signal
|
||||
setupSignals(socket)
|
||||
|
||||
listener, err := setupSocket(socket)
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to setup socket: %v", err)
|
||||
}
|
||||
|
||||
plugin, err := NewPlugin()
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to create a plugin: %v", err)
|
||||
}
|
||||
|
||||
// Cache
|
||||
trafficControlStatusCache = make(map[string]*TrafficControlStatus)
|
||||
|
||||
trafficControlServeMux := http.NewServeMux()
|
||||
|
||||
// Report request handler
|
||||
reportHandler := http.HandlerFunc(plugin.report)
|
||||
trafficControlServeMux.Handle("/report", reportHandler)
|
||||
|
||||
// Control request handler
|
||||
controlHandler := http.HandlerFunc(plugin.control)
|
||||
trafficControlServeMux.Handle("/control", controlHandler)
|
||||
|
||||
log.Println("Listening...")
|
||||
if err = http.Serve(listener, trafficControlServeMux); err != nil {
|
||||
log.Fatalf("failed to serve: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func setupSignals(socket string) {
|
||||
interrupt := make(chan os.Signal, 1)
|
||||
signal.Notify(interrupt, os.Interrupt)
|
||||
go func() {
|
||||
<-interrupt
|
||||
os.Remove(socket)
|
||||
os.Exit(0)
|
||||
}()
|
||||
}
|
||||
|
||||
func setupSocket(socket string) (net.Listener, error) {
|
||||
os.Remove(socket)
|
||||
if err := os.MkdirAll(filepath.Dir(socket), 0755); err != nil {
|
||||
return nil, fmt.Errorf("failed to create directory %q: %v", filepath.Dir(socket), err)
|
||||
}
|
||||
listener, err := net.Listen("unix", socket)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to listen on %q: %v", socket, err)
|
||||
}
|
||||
|
||||
log.Printf("Listening on: unix://%s", socket)
|
||||
return listener, nil
|
||||
}
|
||||
|
||||
// NewPlugin instantiates a new plugin
|
||||
func NewPlugin() (*Plugin, error) {
|
||||
store := NewStore()
|
||||
dockerClient, err := NewDockerClient(store)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create a docker client: %v", err)
|
||||
}
|
||||
reporter := NewReporter(store)
|
||||
plugin := &Plugin{
|
||||
reporter: reporter,
|
||||
clients: []containerClient{
|
||||
dockerClient,
|
||||
},
|
||||
}
|
||||
for _, client := range plugin.clients {
|
||||
go client.Start()
|
||||
}
|
||||
return plugin, nil
|
||||
}
|
||||
|
||||
func (p *Plugin) report(w http.ResponseWriter, r *http.Request) {
|
||||
raw, err := p.reporter.RawReport()
|
||||
if err != nil {
|
||||
msg := fmt.Sprintf("error: failed to get raw report: %v", err)
|
||||
log.Print(msg)
|
||||
http.Error(w, msg, http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write(raw)
|
||||
}
|
||||
|
||||
type request struct {
|
||||
NodeID string
|
||||
Control string
|
||||
}
|
||||
|
||||
type response struct {
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
func (p *Plugin) control(w http.ResponseWriter, r *http.Request) {
|
||||
xreq := request{}
|
||||
if err := json.NewDecoder(r.Body).Decode(&xreq); err != nil {
|
||||
log.Printf("Bad request: %v", err)
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
handler, err := p.reporter.GetHandler(xreq.NodeID, xreq.Control)
|
||||
if err != nil {
|
||||
sendResponse(w, fmt.Errorf("failed to get handler: %v", err))
|
||||
return
|
||||
}
|
||||
if err := handler(); err != nil {
|
||||
sendResponse(w, fmt.Errorf("handler failed: %v", err))
|
||||
return
|
||||
}
|
||||
sendResponse(w, nil)
|
||||
}
|
||||
|
||||
func sendResponse(w http.ResponseWriter, err error) {
|
||||
res := response{}
|
||||
if err != nil {
|
||||
res.Error = err.Error()
|
||||
}
|
||||
raw, err := json.Marshal(res)
|
||||
if err != nil {
|
||||
log.Printf("Internal server error: %v", err)
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
w.Write(raw)
|
||||
}
|
||||
@@ -1,326 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
trafficControlTablePrefix = "traffic-control-table-"
|
||||
)
|
||||
|
||||
type report struct {
|
||||
Container topology
|
||||
Plugins []pluginSpec
|
||||
}
|
||||
|
||||
type topology struct {
|
||||
Nodes map[string]node `json:"nodes"`
|
||||
Controls map[string]control `json:"controls"`
|
||||
MetadataTemplates map[string]metadataTemplate `json:"metadata_templates,omitempty"`
|
||||
TableTemplates map[string]tableTemplate `json:"table_templates,omitempty"`
|
||||
}
|
||||
|
||||
type tableTemplate struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
Prefix string `json:"prefix"`
|
||||
}
|
||||
|
||||
type metadataTemplate struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label,omitempty"` // Human-readable descriptor for this row
|
||||
Truncate int `json:"truncate,omitempty"` // If > 0, truncate the value to this length.
|
||||
Datatype string `json:"dataType,omitempty"`
|
||||
Priority float64 `json:"priority,omitempty"`
|
||||
From string `json:"from,omitempty"` // Defines how to get the value from a report node
|
||||
}
|
||||
|
||||
type node struct {
|
||||
LatestControls map[string]controlEntry `json:"latestControls,omitempty"`
|
||||
Latest map[string]stringEntry `json:"latest,omitempty"`
|
||||
}
|
||||
|
||||
type controlEntry struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Value controlData `json:"value"`
|
||||
}
|
||||
|
||||
type controlData struct {
|
||||
Dead bool `json:"dead"`
|
||||
}
|
||||
|
||||
type control struct {
|
||||
ID string `json:"id"`
|
||||
Human string `json:"human"`
|
||||
Icon string `json:"icon"`
|
||||
Rank int `json:"rank"`
|
||||
}
|
||||
|
||||
type stringEntry struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Value string `json:"value"`
|
||||
}
|
||||
|
||||
type pluginSpec struct {
|
||||
ID string `json:"id"`
|
||||
Label string `json:"label"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Interfaces []string `json:"interfaces"`
|
||||
APIVersion string `json:"api_version,omitempty"`
|
||||
}
|
||||
|
||||
// Reporter internal data structure
|
||||
type Reporter struct {
|
||||
store *Store
|
||||
}
|
||||
|
||||
// NewReporter instantiates a new Reporter
|
||||
func NewReporter(store *Store) *Reporter {
|
||||
return &Reporter{
|
||||
store: store,
|
||||
}
|
||||
}
|
||||
|
||||
// RawReport returns a report
|
||||
func (r *Reporter) RawReport() ([]byte, error) {
|
||||
rpt := &report{
|
||||
Container: topology{
|
||||
Nodes: r.getContainerNodes(),
|
||||
Controls: getTrafficControls(),
|
||||
MetadataTemplates: getMetadataTemplate(),
|
||||
TableTemplates: getTableTemplate(),
|
||||
},
|
||||
Plugins: []pluginSpec{
|
||||
{
|
||||
ID: "traffic-control",
|
||||
Label: "Traffic control",
|
||||
Description: "Adds traffic controls to the running Docker containers",
|
||||
Interfaces: []string{"reporter", "controller"},
|
||||
APIVersion: "1",
|
||||
},
|
||||
},
|
||||
}
|
||||
raw, err := json.Marshal(rpt)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal the report: %v", err)
|
||||
}
|
||||
return raw, nil
|
||||
}
|
||||
|
||||
// GetHandler returns the function performing the action specified by controlID
|
||||
func (r *Reporter) GetHandler(nodeID, controlID string) (func() error, error) {
|
||||
containerID, err := nodeIDToContainerID(nodeID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get container ID from node ID %q: %v", nodeID, err)
|
||||
}
|
||||
container, found := r.store.Container(containerID)
|
||||
if !found {
|
||||
return nil, fmt.Errorf("container %s not found", containerID)
|
||||
}
|
||||
var handler func(pid int) error
|
||||
for _, c := range getControls() {
|
||||
if c.control.ID == controlID {
|
||||
handler = c.handler
|
||||
break
|
||||
}
|
||||
}
|
||||
if handler == nil {
|
||||
return nil, fmt.Errorf("unknown control ID %q for node ID %q", controlID, nodeID)
|
||||
}
|
||||
return func() error {
|
||||
return handler(container.PID)
|
||||
}, nil
|
||||
}
|
||||
|
||||
// states:
|
||||
// created, destroyed - don't create any node
|
||||
// running, not running - create node with controls
|
||||
func (r *Reporter) getContainerNodes() map[string]node {
|
||||
nodes := map[string]node{}
|
||||
timestamp := time.Now()
|
||||
r.store.ForEach(func(containerID string, container Container) {
|
||||
dead := false
|
||||
switch container.State {
|
||||
case Created, Destroyed:
|
||||
// do nothing, to prevent adding a stale node
|
||||
// to a report
|
||||
case Stopped:
|
||||
dead = true
|
||||
fallthrough
|
||||
case Running:
|
||||
nodeID := containerIDToNodeID(containerID)
|
||||
latency, _ := getLatency(container.PID)
|
||||
packetLoss, _ := getPacketLoss(container.PID)
|
||||
nodes[nodeID] = node{
|
||||
LatestControls: getTrafficNodeControls(timestamp, dead),
|
||||
Latest: map[string]stringEntry{
|
||||
fmt.Sprintf("%s%s", trafficControlTablePrefix, "latency"): {
|
||||
Timestamp: timestamp,
|
||||
Value: latency,
|
||||
},
|
||||
fmt.Sprintf("%s%s", trafficControlTablePrefix, "pktloss"): {
|
||||
Timestamp: timestamp,
|
||||
Value: packetLoss,
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
})
|
||||
return nodes
|
||||
}
|
||||
|
||||
func getMetadataTemplate() map[string]metadataTemplate {
|
||||
return map[string]metadataTemplate{
|
||||
"traffic-control-latency": {
|
||||
ID: "traffic-control-latency",
|
||||
Label: "Latency",
|
||||
Truncate: 0,
|
||||
Datatype: "",
|
||||
Priority: 13.5,
|
||||
From: "latest",
|
||||
},
|
||||
"traffic-control-pktloss": {
|
||||
ID: "traffic-control-pktloss",
|
||||
Label: "Packet Loss",
|
||||
Truncate: 0,
|
||||
Datatype: "",
|
||||
Priority: 13.6,
|
||||
From: "latest",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getTableTemplate() map[string]tableTemplate {
|
||||
return map[string]tableTemplate{
|
||||
"traffic-control-table": {
|
||||
ID: "traffic-control-table",
|
||||
Label: "Traffic Control",
|
||||
Prefix: trafficControlTablePrefix,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getTrafficNodeControls(timestamp time.Time, dead bool) map[string]controlEntry {
|
||||
controls := map[string]controlEntry{}
|
||||
entry := controlEntry{
|
||||
Timestamp: timestamp,
|
||||
Value: controlData{
|
||||
Dead: dead,
|
||||
},
|
||||
}
|
||||
for _, c := range getControls() {
|
||||
controls[c.control.ID] = entry
|
||||
}
|
||||
return controls
|
||||
}
|
||||
|
||||
func getTrafficControls() map[string]control {
|
||||
controls := map[string]control{}
|
||||
for _, c := range getControls() {
|
||||
controls[c.control.ID] = c.control
|
||||
}
|
||||
return controls
|
||||
}
|
||||
|
||||
type extControl struct {
|
||||
control control
|
||||
handler func(pid int) error
|
||||
}
|
||||
|
||||
func getLatencyControls() []extControl {
|
||||
return []extControl{
|
||||
{
|
||||
control: control{
|
||||
ID: fmt.Sprintf("%s%s", trafficControlTablePrefix, "slow"),
|
||||
Human: "Traffic speed: slow",
|
||||
Icon: "fa-hourglass-1",
|
||||
Rank: 20,
|
||||
},
|
||||
handler: func(pid int) error {
|
||||
return ApplyLatency(pid, "2000ms")
|
||||
},
|
||||
},
|
||||
{
|
||||
control: control{
|
||||
ID: fmt.Sprintf("%s%s", trafficControlTablePrefix, "medium"),
|
||||
Human: "Traffic speed: medium",
|
||||
Icon: "fa-hourglass-2",
|
||||
Rank: 21,
|
||||
},
|
||||
handler: func(pid int) error {
|
||||
return ApplyLatency(pid, "1000ms")
|
||||
},
|
||||
},
|
||||
{
|
||||
control: control{
|
||||
ID: fmt.Sprintf("%s%s", trafficControlTablePrefix, "fast"),
|
||||
Human: "Traffic speed: fast",
|
||||
Icon: "fa-hourglass-3",
|
||||
Rank: 22,
|
||||
},
|
||||
handler: func(pid int) error {
|
||||
return ApplyLatency(pid, "500ms")
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getPacketLossControls() []extControl {
|
||||
return []extControl{
|
||||
{
|
||||
control: control{
|
||||
ID: fmt.Sprintf("%s%s", trafficControlTablePrefix, "pkt-drop-low"),
|
||||
Human: "Packet drop: low",
|
||||
Icon: "fa-cut",
|
||||
Rank: 23,
|
||||
},
|
||||
handler: func(pid int) error {
|
||||
return ApplyPacketLoss(pid, "10%")
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getGeneralControls() []extControl {
|
||||
return []extControl{
|
||||
{
|
||||
control: control{
|
||||
ID: fmt.Sprintf("%s%s", trafficControlTablePrefix, "clear"),
|
||||
Human: "Clear traffic control settings",
|
||||
Icon: "fa-times-circle",
|
||||
Rank: 24,
|
||||
},
|
||||
handler: func(pid int) error {
|
||||
return ClearTrafficControlSettings(pid)
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func getControls() []extControl {
|
||||
controls := getLatencyControls()
|
||||
// TODO alepuccetti why append(controls, getPacketLossControls()) does not work?
|
||||
for _, ctrl := range getPacketLossControls() {
|
||||
controls = append(controls, ctrl)
|
||||
}
|
||||
for _, ctrl := range getGeneralControls() {
|
||||
controls = append(controls, ctrl)
|
||||
}
|
||||
return controls
|
||||
}
|
||||
|
||||
const nodeSuffix = ";<container>"
|
||||
|
||||
func containerIDToNodeID(containerID string) string {
|
||||
return fmt.Sprintf("%s%s", containerID, nodeSuffix)
|
||||
}
|
||||
|
||||
func nodeIDToContainerID(nodeID string) (string, error) {
|
||||
if !strings.HasSuffix(nodeID, nodeSuffix) {
|
||||
return "", fmt.Errorf("no suffix %q in node ID %q", nodeSuffix, nodeID)
|
||||
}
|
||||
return strings.TrimSuffix(nodeID, nodeSuffix), nil
|
||||
}
|
||||
@@ -1,69 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
// State is the container internal state
|
||||
type State int
|
||||
|
||||
const (
|
||||
// Created state
|
||||
Created State = iota
|
||||
// Running state
|
||||
Running
|
||||
// Stopped state
|
||||
Stopped
|
||||
// Destroyed state
|
||||
Destroyed
|
||||
)
|
||||
|
||||
// Container data structure
|
||||
type Container struct {
|
||||
State State
|
||||
PID int
|
||||
}
|
||||
|
||||
// Store data structure
|
||||
type Store struct {
|
||||
lock sync.Mutex
|
||||
containers map[string]Container
|
||||
}
|
||||
|
||||
// NewStore instantiates a new Store
|
||||
func NewStore() *Store {
|
||||
return &Store{
|
||||
containers: map[string]Container{},
|
||||
}
|
||||
}
|
||||
|
||||
// Container returns a container form its ID
|
||||
func (s *Store) Container(containerID string) (Container, bool) {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
container, found := s.containers[containerID]
|
||||
return container, found
|
||||
}
|
||||
|
||||
// SetContainer sets a container into the store
|
||||
func (s *Store) SetContainer(containerID string, container Container) {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
s.containers[containerID] = container
|
||||
}
|
||||
|
||||
// DeleteContainer deletes a container from the store
|
||||
func (s *Store) DeleteContainer(containerID string) {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
delete(s.containers, containerID)
|
||||
}
|
||||
|
||||
// ForEach execute a function on each container in the store
|
||||
func (s *Store) ForEach(callback func(ID string, c Container)) {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
for containerID, container := range s.containers {
|
||||
callback(containerID, container)
|
||||
}
|
||||
}
|
||||
@@ -1,206 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"strings"
|
||||
|
||||
log "github.com/Sirupsen/logrus"
|
||||
"github.com/containernetworking/cni/pkg/ns"
|
||||
)
|
||||
|
||||
// applyTrafficControlRules set the network policies
|
||||
func applyTrafficControlRules(pid int, rules []string) (netNSID string, err error) {
|
||||
cmds := [][]string{
|
||||
strings.Fields("tc qdisc replace dev eth0 root handle 1: netem"),
|
||||
}
|
||||
cmd := strings.Fields("tc qdisc change dev eth0 root handle 1: netem")
|
||||
cmd = append(cmd, rules...)
|
||||
cmds = append(cmds, cmd)
|
||||
|
||||
netNS := fmt.Sprintf("/proc/%d/ns/net", pid)
|
||||
err = ns.WithNetNSPath(netNS, func(hostNS ns.NetNS) error {
|
||||
for _, cmd := range cmds {
|
||||
if output, err := exec.Command(cmd[0], cmd[1:]...).CombinedOutput(); err != nil {
|
||||
log.Error(string(output))
|
||||
return fmt.Errorf("failed to execute command: %v", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to perform traffic control: %v", err)
|
||||
}
|
||||
netNSID, err = getNSID(netNS)
|
||||
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return netNSID, nil
|
||||
}
|
||||
|
||||
// ApplyLatency sets the latency
|
||||
func ApplyLatency(pid int, latency string) error {
|
||||
if latency == "" {
|
||||
return nil
|
||||
}
|
||||
rules := strings.Fields(fmt.Sprintf("delay %s", latency))
|
||||
|
||||
// Get cached packet loss
|
||||
packetLoss, err := getPacketLoss(pid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if packetLoss != "-" {
|
||||
rules = append(rules, strings.Fields(fmt.Sprintf("loss %s", packetLoss))...)
|
||||
}
|
||||
|
||||
netNSID, err := applyTrafficControlRules(pid, rules)
|
||||
|
||||
// Update cached values
|
||||
if trafficControlStatusCache[netNSID] == nil {
|
||||
trafficControlStatusCache[netNSID] = TrafficControlStatusInit()
|
||||
}
|
||||
trafficControlStatusCache[netNSID].SetLatency(latency)
|
||||
trafficControlStatusCache[netNSID].SetPacketLoss(packetLoss)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ApplyPacketLoss sets the packet loss
|
||||
func ApplyPacketLoss(pid int, packetLoss string) error {
|
||||
if packetLoss == "" {
|
||||
return nil
|
||||
}
|
||||
rules := strings.Fields(fmt.Sprintf("loss %s", packetLoss))
|
||||
|
||||
// Get cached latency
|
||||
latency, err := getLatency(pid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if latency != "-" {
|
||||
rules = append(rules, strings.Fields(fmt.Sprintf("delay %s", latency))...)
|
||||
}
|
||||
|
||||
netNSID, err := applyTrafficControlRules(pid, rules)
|
||||
|
||||
// Update cached values
|
||||
if trafficControlStatusCache[netNSID] == nil {
|
||||
trafficControlStatusCache[netNSID] = TrafficControlStatusInit()
|
||||
}
|
||||
trafficControlStatusCache[netNSID].SetLatency(latency)
|
||||
trafficControlStatusCache[netNSID].SetPacketLoss(packetLoss)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// ClearTrafficControlSettings clear all parameters of the qdisc with tc
|
||||
func ClearTrafficControlSettings(pid int) error {
|
||||
cmds := [][]string{
|
||||
strings.Fields("tc qdisc replace dev eth0 root handle 1: netem"),
|
||||
}
|
||||
netNS := fmt.Sprintf("/proc/%d/ns/net", pid)
|
||||
err := ns.WithNetNSPath(netNS, func(hostNS ns.NetNS) error {
|
||||
for _, cmd := range cmds {
|
||||
if output, err := exec.Command(cmd[0], cmd[1:]...).CombinedOutput(); err != nil {
|
||||
log.Error(string(output))
|
||||
return fmt.Errorf("failed to execute command: %v", err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to perform traffic control: %v", err)
|
||||
}
|
||||
// clear cached parameters
|
||||
netNSID, err := getNSID(netNS)
|
||||
if err != nil {
|
||||
log.Error(netNSID)
|
||||
return fmt.Errorf("failed to get network namespace ID: %v", err)
|
||||
}
|
||||
delete(trafficControlStatusCache, netNSID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func getLatency(pid int) (string, error) {
|
||||
var status *TrafficControlStatus
|
||||
var err error
|
||||
if status, err = getStatus(pid); err != nil {
|
||||
return "-", err
|
||||
} else if status == nil {
|
||||
return "-", fmt.Errorf("status for PID %d does not exist", pid)
|
||||
}
|
||||
return status.latency, nil
|
||||
}
|
||||
|
||||
func getPacketLoss(pid int) (string, error) {
|
||||
var status *TrafficControlStatus
|
||||
var err error
|
||||
if status, err = getStatus(pid); err != nil {
|
||||
return "-", err
|
||||
} else if status == nil {
|
||||
return "-", fmt.Errorf("status for PID %d does not exist", pid)
|
||||
}
|
||||
return status.packetLoss, nil
|
||||
}
|
||||
|
||||
func getStatus(pid int) (*TrafficControlStatus, error) {
|
||||
netNS := fmt.Sprintf("/proc/%d/ns/net", pid)
|
||||
netNSID, err := getNSID(netNS)
|
||||
if err != nil {
|
||||
log.Error(netNSID)
|
||||
return nil, fmt.Errorf("failed to get network namespace ID: %v", err)
|
||||
}
|
||||
if status, ok := trafficControlStatusCache[netNSID]; ok {
|
||||
return status, nil
|
||||
}
|
||||
cmd := strings.Fields("tc qdisc show dev eth0")
|
||||
var output string
|
||||
err = ns.WithNetNSPath(netNS, func(hostNS ns.NetNS) error {
|
||||
cmdOut, err := exec.Command(cmd[0], cmd[1:]...).CombinedOutput()
|
||||
if err != nil {
|
||||
log.Error(string(cmdOut))
|
||||
output = ""
|
||||
return fmt.Errorf("failed to execute command: tc qdisc show dev eth0: %v", err)
|
||||
}
|
||||
output = string(cmdOut)
|
||||
return nil
|
||||
})
|
||||
// cache parameters
|
||||
trafficControlStatusCache[netNSID] = &TrafficControlStatus{
|
||||
latency: parseLatency(output),
|
||||
packetLoss: parsePacketLoss(output),
|
||||
}
|
||||
status, _ := trafficControlStatusCache[netNSID]
|
||||
return status, err
|
||||
}
|
||||
|
||||
func parseLatency(statusString string) string {
|
||||
return parseAttribute(statusString, "delay")
|
||||
}
|
||||
|
||||
func parsePacketLoss(statusString string) string {
|
||||
return parseAttribute(statusString, "loss")
|
||||
}
|
||||
func parseAttribute(statusString string, attribute string) string {
|
||||
statusStringSplited := strings.Fields(statusString)
|
||||
for i, s := range statusStringSplited {
|
||||
if s == attribute {
|
||||
if i < len(statusStringSplited)-1 {
|
||||
return strings.Trim(statusStringSplited[i+1], "\n")
|
||||
}
|
||||
return "-"
|
||||
}
|
||||
}
|
||||
return "-"
|
||||
}
|
||||
|
||||
func getNSID(nsPath string) (string, error) {
|
||||
nsID, err := os.Readlink(nsPath)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed read \"%s\": %v", nsPath, err)
|
||||
}
|
||||
return nsID[5 : len(nsID)-1], nil
|
||||
}
|
||||
Reference in New Issue
Block a user