src: Add DMA localagent

Change-Id: Ibcee814fbc9a904448eeb368a1a26bbb69cf54aa
Signed-off-by: Toshiaki Takahashi <takahashi.tsc@ncos.nec.co.jp>

11 files changed, 1521 insertions, 0 deletions

diff --git a/src/dma/cmd/infofetch/daemon.go b/src/dma/cmd/infofetch/daemon.go
new file mode 100644
index 00000000..d4ff94f5
--- /dev/null
+++ b/src/dma/cmd/infofetch/daemon.go
@@ -0,0 +1,103 @@
+/*
+ * Copyright 2018 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"github.com/BurntSushi/toml"
+	"github.com/distributed-monitoring/agent/pkg/common"
+	"github.com/go-redis/redis"
+	libvirt "github.com/libvirt/libvirt-go"
+	"log"
+	"sync"
+)
+
+var infoPool common.RedisPool
+
+// Config is ...
+type Config struct {
+	Common    CommonConfig
+	InfoFetch InfoFetchConfig
+}
+
+// CommonConfig is ...
+type CommonConfig struct {
+	RedisHost     string `toml:"redis_host"`
+	RedisPort     string `toml:"redis_port"`
+	RedisPassword string `toml:"redis_password"`
+	RedisDB       int    `toml:"redis_db"`
+}
+
+// InfoFetchConfig is ...
+type InfoFetchConfig struct {
+	OSUsername          string `toml:"os_username"`
+	OSUserDomainName    string `toml:"os_user_domain_name"`
+	OSProjectDomainName string `toml:"os_project_domain_name"`
+	OSProjectName       string `toml:"os_project_name"`
+	OSPassword          string `toml:"os_password"`
+	OSAuthURL           string `toml:"os_auth_url"`
+}
+
+func main() {
+
+	var config Config
+	_, err := toml.DecodeFile("/etc/barometer-dma/config.toml", &config)
+	if err != nil {
+		log.Println("Read error of config file")
+	}
+
+	var waitgroup sync.WaitGroup
+	libvirt.EventRegisterDefaultImpl()
+
+	redisClient := redis.NewClient(&redis.Options{
+		Addr:     config.Common.RedisHost + ":" + config.Common.RedisPort,
+		Password: config.Common.RedisPassword,
+		DB:       config.Common.RedisDB,
+	})
+	infoPool = common.RedisPool{Client: redisClient}
+	// Initialize redis db...
+	infoPool.DelAll()
+
+	conn, err := libvirt.NewConnect("qemu:///system")
+	if err != nil {
+		log.Fatalln("libvirt connect error")
+	}
+	defer conn.Close()
+
+	vmIfInfoChan := make(chan string)
+	{
+		ctx := context.Background()
+		waitgroup.Add(1)
+		go func() {
+			RunNeutronInfoFetch(ctx, &config, vmIfInfoChan)
+			waitgroup.Done()
+		}()
+	}
+
+	//Get active VM info
+	GetActiveDomain(conn, vmIfInfoChan)
+	{
+		ctx := context.Background()
+		waitgroup.Add(1)
+		go func() {
+			RunVirshEventLoop(ctx, conn, vmIfInfoChan)
+			waitgroup.Done()
+		}()
+	}
+
+	waitgroup.Wait()
+}
diff --git a/src/dma/cmd/infofetch/openstack.go b/src/dma/cmd/infofetch/openstack.go
new file mode 100644
index 00000000..c0c54f5a
--- /dev/null
+++ b/src/dma/cmd/infofetch/openstack.go
@@ -0,0 +1,522 @@
+/*
+ * Copyright 2017 Red Hat
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"text/template"
+	"time"
+)
+
+var env *InfoFetchConfig
+
+type userInfo struct {
+	UserDomainName    string
+	UserName          string
+	Password          string
+	ProjectDomainName string
+	ProjectName       string
+}
+
+var tokenJSONTemplate = `{
+  "auth": {
+    "identity": {
+      "methods": [
+        "password"
+      ],
+      "password": {
+        "user": {
+          "domain": {
+            "name": "{{.UserDomainName}}"
+          },
+          "name": "{{.UserName}}",
+          "password": "{{.Password}}"
+        }
+      }
+    },
+    "scope": {
+      "project": {
+        "domain": {
+          "name": "{{.ProjectDomainName}}"
+        },
+        "name": "{{.ProjectName}}"
+      }
+    }
+  }
+}
+`
+
+type tokenReply struct {
+	Token struct {
+		IsDomain bool     `json:"is_domain"`
+		Methods  []string `json:"methods"`
+		Roles    []struct {
+			ID   string `json:"id"`
+			Name string `json:"name"`
+		} `json:"roles"`
+		ExpiresAt time.Time `json:"expires_at"`
+		Project   struct {
+			Domain struct {
+				ID   string `json:"id"`
+				Name string `json:"name"`
+			} `json:"domain"`
+			ID   string `json:"id"`
+			Name string `json:"name"`
+		} `json:"project"`
+		User struct {
+			PasswordExpiresAt interface{} `json:"password_expires_at"`
+			Domain            struct {
+				ID   string `json:"id"`
+				Name string `json:"name"`
+			} `json:"domain"`
+			ID   string `json:"id"`
+			Name string `json:"name"`
+		} `json:"user"`
+		AuditIds []string  `json:"audit_ids"`
+		IssuedAt time.Time `json:"issued_at"`
+	} `json:"token"`
+}
+
+type token struct {
+	Token     string
+	ExpiresAt time.Time
+}
+
+func (t *token) CheckToken() {
+	now := time.Now()
+
+	if t.ExpiresAt.Sub(now).Seconds() < 30 {
+		newToken, _ := getToken()
+		t.Token = newToken.Token
+		t.ExpiresAt = newToken.ExpiresAt
+	}
+}
+
+func getToken() (*token, error) {
+	var buf bytes.Buffer
+
+	t := template.Must(template.New("json template1").Parse(tokenJSONTemplate))
+	p := userInfo{
+		UserDomainName:    env.OSUserDomainName,
+		UserName:          env.OSUsername,
+		Password:          env.OSPassword,
+		ProjectDomainName: env.OSProjectDomainName,
+		ProjectName:       env.OSProjectName,
+	}
+	t.Execute(&buf, p)
+
+	body := bytes.NewReader(buf.Bytes())
+	req, err := http.NewRequest("POST", env.OSAuthURL+"/auth/tokens?nocatalog", body)
+	if err != nil {
+		return &token{"", time.Unix(0, 0)}, fmt.Errorf("http request failed: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return &token{"", time.Unix(0, 0)}, fmt.Errorf("http POST failed: %v", err)
+	}
+	defer resp.Body.Close()
+	b, err := ioutil.ReadAll(resp.Body)
+
+	tokenStr, ok := resp.Header["X-Subject-Token"]
+	if ok != true && len(tokenStr) != 1 {
+		return &token{"", time.Unix(0, 0)}, fmt.Errorf("no token in openstack reply")
+	}
+
+	var repl tokenReply
+	err = json.Unmarshal(b, &repl)
+
+	return &token{tokenStr[0], repl.Token.ExpiresAt}, nil
+}
+
+type service struct {
+	Description string `json:"description"`
+	Links       struct {
+		Self string `json:"self"`
+	} `json:"links"`
+	Enabled bool   `json:"enabled"`
+	Type    string `json:"type"`
+	ID      string `json:"id"`
+	Name    string `json:"name"`
+}
+
+type serviceListReply struct {
+	Services []service `json:"services"`
+}
+
+func (s *serviceListReply) GetService(name string) (*service, error) {
+	for _, v := range s.Services {
+		if v.Name == name {
+			return &v, nil
+		}
+	}
+	return nil, fmt.Errorf("no service id (%s) found", name)
+}
+
+type endPoint struct {
+	RegionID string `json:"region_id"`
+	Links    struct {
+		Self string `json:"self"`
+	} `json:"links"`
+	URL       string `json:"url"`
+	Region    string `json:"region"`
+	Enabled   bool   `json:"enabled"`
+	Interface string `json:"interface"`
+	ServiceID string `json:"service_id"`
+	ID        string `json:"id"`
+}
+
+type endPointReply struct {
+	Endpoints []endPoint `json:"endpoints"`
+}
+
+func (e *endPointReply) GetEndpoint(serviceid string, ifname string) (*endPoint, error) {
+	for _, v := range e.Endpoints {
+		if v.Interface == ifname && v.ServiceID == serviceid {
+			return &v, nil
+		}
+	}
+	return nil, fmt.Errorf("no endpoint found (%s/%s)", serviceid, ifname)
+}
+
+func getEndpoints(token *token) (endPointReply, error) {
+	token.CheckToken()
+	req, err := http.NewRequest("GET", env.OSAuthURL+"/endpoints", nil)
+	if err != nil {
+		return endPointReply{}, fmt.Errorf("request failed:%v", err)
+	}
+	req.Header.Set("X-Auth-Token", token.Token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return endPointReply{}, fmt.Errorf("http GET failed:%v", err)
+	}
+	defer resp.Body.Close()
+
+	b, err := ioutil.ReadAll(resp.Body)
+	//fmt.Printf("%s", string(b))
+
+	var repl endPointReply
+	err = json.Unmarshal(b, &repl)
+	if err != nil {
+		return endPointReply{}, fmt.Errorf("http reply decoding failed:%v", err)
+	}
+	//fmt.Printf("%v", repl)
+	return repl, nil
+}
+
+func getServiceList(token *token) (serviceListReply, error) {
+	token.CheckToken()
+	req, err := http.NewRequest("GET", env.OSAuthURL+"/services", nil)
+	if err != nil {
+		return serviceListReply{}, fmt.Errorf("request failed:%v", err)
+	}
+	req.Header.Set("X-Auth-Token", token.Token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return serviceListReply{}, fmt.Errorf("http GET failed:%v", err)
+	}
+	defer resp.Body.Close()
+	b, err := ioutil.ReadAll(resp.Body)
+
+	var repl serviceListReply
+	err = json.Unmarshal(b, &repl)
+	if err != nil {
+		return serviceListReply{}, fmt.Errorf("http reply decoding failed:%v", err)
+	}
+	return repl, nil
+}
+
+type neutronPort struct {
+	AllowedAddressPairs []interface{} `json:"allowed_address_pairs"`
+	ExtraDhcpOpts       []interface{} `json:"extra_dhcp_opts"`
+	UpdatedAt           time.Time     `json:"updated_at"`
+	DeviceOwner         string        `json:"device_owner"`
+	RevisionNumber      int           `json:"revision_number"`
+	PortSecurityEnabled bool          `json:"port_security_enabled"`
+	BindingProfile      struct {
+	} `json:"binding:profile"`
+	FixedIps []struct {
+		SubnetID  string `json:"subnet_id"`
+		IPAddress string `json:"ip_address"`
+	} `json:"fixed_ips"`
+	ID                string        `json:"id"`
+	SecurityGroups    []interface{} `json:"security_groups"`
+	BindingVifDetails struct {
+		PortFilter    bool   `json:"port_filter"`
+		DatapathType  string `json:"datapath_type"`
+		OvsHybridPlug bool   `json:"ovs_hybrid_plug"`
+	} `json:"binding:vif_details"`
+	BindingVifType  string        `json:"binding:vif_type"`
+	MacAddress      string        `json:"mac_address"`
+	ProjectID       string        `json:"project_id"`
+	Status          string        `json:"status"`
+	BindingHostID   string        `json:"binding:host_id"`
+	Description     string        `json:"description"`
+	Tags            []interface{} `json:"tags"`
+	QosPolicyID     interface{}   `json:"qos_policy_id"`
+	Name            string        `json:"name"`
+	AdminStateUp    bool          `json:"admin_state_up"`
+	NetworkID       string        `json:"network_id"`
+	TenantID        string        `json:"tenant_id"`
+	CreatedAt       time.Time     `json:"created_at"`
+	BindingVnicType string        `json:"binding:vnic_type"`
+	DeviceID        string        `json:"device_id"`
+}
+
+type neutronPortReply struct {
+	Ports []neutronPort `json:"ports"`
+}
+
+func getNeutronPorts(token *token, endpoint string) (neutronPortReply, error) {
+	token.CheckToken()
+	req, err := http.NewRequest("GET", endpoint+"/v2.0/ports", nil)
+	if err != nil {
+		return neutronPortReply{}, fmt.Errorf("request failed:%v", err)
+	}
+	req.Header.Set("X-Auth-Token", token.Token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return neutronPortReply{}, fmt.Errorf("http GET failed:%v", err)
+	}
+	defer resp.Body.Close()
+
+	b, err := ioutil.ReadAll(resp.Body)
+
+	var repl neutronPortReply
+	err = json.Unmarshal(b, &repl)
+	if err != nil {
+		return neutronPortReply{}, fmt.Errorf("http reply decoding failed:%v", err)
+	}
+	return repl, nil
+}
+
+func (n *neutronPortReply) GetNeutronPortfromMAC(mac string) (*neutronPort,
+	error) {
+	for _, v := range n.Ports {
+		if v.MacAddress == strings.ToLower(mac) {
+			return &v, nil
+		}
+	}
+	return nil, fmt.Errorf("no port (%s) found", mac)
+}
+
+type neutronNetwork struct {
+	ProviderPhysicalNetwork string        `json:"provider:physical_network"`
+	Ipv6AddressScope        interface{}   `json:"ipv6_address_scope"`
+	RevisionNumber          int           `json:"revision_number"`
+	PortSecurityEnabled     bool          `json:"port_security_enabled"`
+	Mtu                     int           `json:"mtu"`
+	ID                      string        `json:"id"`
+	RouterExternal          bool          `json:"router:external"`
+	AvailabilityZoneHints   []interface{} `json:"availability_zone_hints"`
+	AvailabilityZones       []string      `json:"availability_zones"`
+	ProviderSegmentationID  interface{}   `json:"provider:segmentation_id"`
+	Ipv4AddressScope        interface{}   `json:"ipv4_address_scope"`
+	Shared                  bool          `json:"shared"`
+	ProjectID               string        `json:"project_id"`
+	Status                  string        `json:"status"`
+	Subnets                 []string      `json:"subnets"`
+	Description             string        `json:"description"`
+	Tags                    []interface{} `json:"tags"`
+	UpdatedAt               time.Time     `json:"updated_at"`
+	IsDefault               bool          `json:"is_default"`
+	QosPolicyID             interface{}   `json:"qos_policy_id"`
+	Name                    string        `json:"name"`
+	AdminStateUp            bool          `json:"admin_state_up"`
+	TenantID                string        `json:"tenant_id"`
+	CreatedAt               time.Time     `json:"created_at"`
+	ProviderNetworkType     string        `json:"provider:network_type"`
+}
+
+type neutronNetworkReply struct {
+	Networks []neutronNetwork `json:"networks"`
+}
+
+func (n *neutronNetworkReply) GetNetworkFromID(netid string) (*neutronNetwork, error) {
+	for _, v := range n.Networks {
+		if v.ID == netid {
+			return &v, nil
+		}
+	}
+	return nil, fmt.Errorf("no network (%s) found", netid)
+}
+
+func getNetworkReply(token *token, endpoint string) (neutronNetworkReply, error) {
+	token.CheckToken()
+	req, err := http.NewRequest("GET", endpoint+"/v2.0/networks", nil)
+	if err != nil {
+		return neutronNetworkReply{}, fmt.Errorf("request failed:%v", err)
+	}
+	req.Header.Set("X-Auth-Token", token.Token)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return neutronNetworkReply{}, fmt.Errorf("http GET failed:%v", err)
+	}
+	defer resp.Body.Close()
+
+	b, err := ioutil.ReadAll(resp.Body)
+
+	var repl neutronNetworkReply
+	err = json.Unmarshal(b, &repl)
+	if err != nil {
+		return neutronNetworkReply{}, fmt.Errorf("http reply decoding failed:%v", err)
+	}
+	return repl, nil
+}
+
+type novaCompute struct {
+	ID    string `json:"id"`
+	Links []struct {
+		Href string `json:"href"`
+		Rel  string `json:"rel"`
+	} `json:"links"`
+	Name string `json:"name"`
+}
+
+type novaComputeReply struct {
+	Servers []novaCompute `json:"servers"`
+}
+
+func (n *novaComputeReply) GetComputeFromID(vmid string) (*novaCompute, error) {
+	for _, v := range n.Servers {
+		if v.ID == vmid {
+			return &v, nil
+		}
+	}
+	return nil, fmt.Errorf("no vm (%s) found", vmid)
+}
+
+func getComputeReply(token *token, endpoint string) (novaComputeReply, error) {
+	token.CheckToken()
+	values := url.Values{}
+	values.Add("all_tenants", "1")
+
+	req, err := http.NewRequest("GET", endpoint+"/servers", nil)
+	if err != nil {
+		return novaComputeReply{}, fmt.Errorf("request failed:%v", err)
+	}
+	req.Header.Set("X-Auth-Token", token.Token)
+	req.URL.RawQuery = values.Encode()
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return novaComputeReply{}, fmt.Errorf("http GET failed:%v", err)
+	}
+	defer resp.Body.Close()
+
+	b, err := ioutil.ReadAll(resp.Body)
+
+	var repl novaComputeReply
+	err = json.Unmarshal(b, &repl)
+	if err != nil {
+		return novaComputeReply{}, fmt.Errorf("http reply decoding failed:%v", err)
+	}
+
+	return repl, nil
+}
+
+type osNeutronInterfaceAnnotation struct {
+	IfName      string
+	VMName      string
+	NetworkName string
+}
+
+// RunNeutronInfoFetch gets redis key update from libvirt and get network information
+// from Neutron with REST api. The retrieved information is stored under redis,
+// if/<tap name>/neutron_network
+func RunNeutronInfoFetch(ctx context.Context, config *Config, vmIfInfo chan string) error {
+	env = &config.InfoFetch
+	token, err := getToken()
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "cannot get token: %v", err)
+		return err
+	}
+
+	svc, _ := getServiceList(token)
+	neuID, _ := svc.GetService("neutron")
+	//fmt.Printf("neutron id:%s\n", id.ID)
+
+	novaID, _ := svc.GetService("nova")
+	//fmt.Printf("nova id:%s\n", id.ID)
+
+	endpoints, _ := getEndpoints(token)
+	neuEndpoint, _ := endpoints.GetEndpoint(neuID.ID, "admin")
+	//fmt.Printf("neutron endpoint:%s\n", neuEndpoint.URL)
+
+	novaEndpoint, _ := endpoints.GetEndpoint(novaID.ID, "admin")
+	//fmt.Printf("nova endpoint:%s\n", novaEndpoint.URL)
+
+	getComputeReply(token, novaEndpoint.URL)
+	getNeutronPorts(token, neuEndpoint.URL)
+	//vmrepl, _ := getComputeReply(token, novaEndpoint.URL)
+	//prepl, _ := getNeutronPorts(token, neuEndpoint.URL)
+
+EVENTLOOP:
+	for {
+		select {
+		case <-ctx.Done():
+			break EVENTLOOP
+		case key := <-vmIfInfo:
+			log.Printf("Incoming IF: %v", key)
+			libvirtIfInfo, err := infoPool.Get(key)
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			} else {
+				var ifInfo osVMInterfaceAnnotation
+				err = json.Unmarshal([]byte(libvirtIfInfo), &ifInfo)
+				if err != nil {
+					log.Fatalf("Err: %v", err)
+				} else {
+					vmrepl, _ := getComputeReply(token, novaEndpoint.URL)
+					prepl, _ := getNeutronPorts(token, neuEndpoint.URL)
+					nrepl, _ := getNetworkReply(token, neuEndpoint.URL)
+					netid, _ := prepl.GetNeutronPortfromMAC(ifInfo.MacAddr)
+					net, _ := nrepl.GetNetworkFromID(netid.NetworkID)
+					vm, _ := vmrepl.GetComputeFromID(netid.DeviceID)
+					osIfInfo := osNeutronInterfaceAnnotation{
+						IfName:      ifInfo.Target,
+						VMName:      vm.Name,
+						NetworkName: net.Name}
+
+					osIfInfoJSON, err := json.Marshal(osIfInfo)
+					if err != nil {
+						log.Fatalf("Err: %v", err)
+					} else {
+						log.Printf("Get: vmname: %s / networkname:%s", vm.Name, net.Name)
+						infoPool.Set(fmt.Sprintf("if/%s/%s", ifInfo.Target, "neutron_network"), string(osIfInfoJSON))
+					}
+				}
+			}
+		}
+	}
+	return nil
+}
diff --git a/src/dma/cmd/infofetch/virsh_domain.go b/src/dma/cmd/infofetch/virsh_domain.go
new file mode 100644
index 00000000..b79f5bdd
--- /dev/null
+++ b/src/dma/cmd/infofetch/virsh_domain.go
@@ -0,0 +1,264 @@
+/*
+ * Copyright 2017 Red Hat
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"encoding/xml"
+	"fmt"
+	libvirt "github.com/libvirt/libvirt-go"
+	"log"
+)
+
+type instance struct {
+	Name  string `xml:"name"`
+	Owner struct {
+		User    string `xml:"user"`
+		Project string `xml:"project"`
+	} `xml:"owner"`
+	Flavor struct {
+		Name string `xml:"name,attr"`
+	} `xml:"flavor"`
+}
+
+type domain struct {
+	Name    string `xml:"name"`
+	Devices struct {
+		Interfaces []struct {
+			Type string `xml:"type,attr"`
+			Mac  struct {
+				Address string `xml:"address,attr"`
+			} `xml:"mac"`
+			Target struct {
+				Dev string `xml:"dev,attr"`
+			} `xml:"target"`
+		} `xml:"interface"`
+	} `xml:"devices"`
+}
+
+type osVMAnnotation struct {
+	Name    string
+	Owner   string
+	Project string
+	Flavor  string
+}
+
+type osVMInterfaceAnnotation struct {
+	Type    string
+	MacAddr string
+	Target  string
+	VMName  string
+}
+
+func parseNovaMetadata(metadata string) (*osVMAnnotation, error) {
+	data := new(instance)
+
+	if err := xml.Unmarshal([]byte(metadata), data); err != nil {
+		log.Println("XML Unmarshal error:", err)
+		return nil, err
+	}
+	log.Printf("Get name: %s user: %s, project: %s, flavor: %s", data.Name, data.Owner.User, data.Owner.Project, data.Flavor.Name)
+	return &osVMAnnotation{
+		Name:    data.Name,
+		Owner:   data.Owner.User,
+		Project: data.Owner.Project,
+		Flavor:  data.Flavor.Name}, nil
+}
+
+func parseXMLForMAC(dumpxml string) (*[]osVMInterfaceAnnotation, error) {
+	data := new(domain)
+
+	if err := xml.Unmarshal([]byte(dumpxml), data); err != nil {
+		log.Println("XML Unmarshal error:", err)
+		return nil, err
+	}
+
+	ifAnnotation := make([]osVMInterfaceAnnotation, len(data.Devices.Interfaces))
+	for i, v := range data.Devices.Interfaces {
+		log.Printf("Interface type: %s, mac_addr: %s, target_dev: %s", v.Type, v.Mac.Address, v.Target.Dev)
+		ifAnnotation[i] = osVMInterfaceAnnotation{
+			Type:    v.Type,
+			MacAddr: v.Mac.Address,
+			Target:  v.Target.Dev,
+			VMName:  data.Name}
+	}
+	return &ifAnnotation, nil
+}
+
+func setInterfaceAnnotation(ifInfo *[]osVMInterfaceAnnotation, vmIfInfoChan chan string) {
+	for _, v := range *ifInfo {
+		ifInfoJSON, err := json.Marshal(v)
+		if err != nil {
+			log.Fatalf("Err: %v", err)
+		}
+		infoPool.Set(fmt.Sprintf("if/%s/%s", v.Target, "network"), string(ifInfoJSON))
+
+		vmIfInfoChan <- fmt.Sprintf("if/%s/%s", v.Target, "network")
+	}
+	return
+}
+
+func domainEventLifecycleCallback(vmIfInfo chan string) func(c *libvirt.Connect, d *libvirt.Domain, event *libvirt.DomainEventLifecycle) {
+
+	return func(c *libvirt.Connect,
+		d *libvirt.Domain, event *libvirt.DomainEventLifecycle) {
+		domName, _ := d.GetName()
+
+		switch event.Event {
+		case libvirt.DOMAIN_EVENT_DEFINED:
+			// VM defined: vmname (libvirt, nova), user, project, flavor
+			// Redis: <vnname>/vminfo
+			log.Printf("Event defined: domName: %s, event: %v", domName, event)
+			metadata, err := d.GetMetadata(libvirt.DOMAIN_METADATA_ELEMENT, "http://openstack.org/xmlns/libvirt/nova/1.0", libvirt.DOMAIN_AFFECT_CONFIG)
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			}
+			vmInfo, err := parseNovaMetadata(metadata)
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			}
+			vmInfoJSON, err := json.Marshal(vmInfo)
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			}
+			infoPool.Set(fmt.Sprintf("vm/%s/%s", domName, "vminfo"), string(vmInfoJSON))
+		case libvirt.DOMAIN_EVENT_STARTED:
+			// VM started: interface type, interface mac addr, intarface type
+			// Redis: <vnname>/interfaces
+			log.Printf("Event started: domName: %s, event: %v", domName, event)
+
+			xml, err := d.GetXMLDesc(0)
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			}
+			ifInfo, err := parseXMLForMAC(xml)
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			}
+			setInterfaceAnnotation(ifInfo, vmIfInfo)
+
+			ifInfoJSON, err := json.Marshal(ifInfo)
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			}
+			infoPool.Set(fmt.Sprintf("vm/%s/%s", domName, "interfaces"), string(ifInfoJSON))
+		case libvirt.DOMAIN_EVENT_UNDEFINED:
+			log.Printf("Event undefined: domName: %s, event: %v", domName, event)
+			vmIFInfo, err := infoPool.Get(fmt.Sprintf("vm/%s/%s", domName, "interfaces"))
+			if err != nil {
+				log.Fatalf("Err: %v", err)
+			} else {
+				var interfaces []osVMInterfaceAnnotation
+				err = json.Unmarshal([]byte(vmIFInfo), &interfaces)
+				if err != nil {
+					log.Fatalf("Err: %v", err)
+				} else {
+					for _, v := range interfaces {
+						infoPool.Del(fmt.Sprintf("if/%s/%s", v.Target, "network"))
+						infoPool.Del(fmt.Sprintf("if/%s/%s", v.Target, "neutron_network"))
+					}
+				}
+			}
+			infoPool.Del(fmt.Sprintf("vm/%s/%s", domName, "vminfo"))
+			infoPool.Del(fmt.Sprintf("vm/%s/%s", domName, "interfaces"))
+		default:
+			log.Printf("Event misc: domName: %s, event: %v", domName, event)
+		}
+	}
+}
+
+// GetActiveDomain gets all active domain information from libvirt and it should be called at startup to get
+// current running domain information
+func GetActiveDomain(conn *libvirt.Connect, vmIfInfoChan chan string) error {
+	doms, err := conn.ListAllDomains(libvirt.CONNECT_LIST_DOMAINS_ACTIVE)
+	if err != nil {
+		log.Fatalf("libvirt dom list error: %s", err)
+		return err
+	}
+
+	for _, d := range doms {
+		name, err := d.GetName()
+
+		// Get VM Info
+		metadata, err := d.GetMetadata(libvirt.DOMAIN_METADATA_ELEMENT, "http://openstack.org/xmlns/libvirt/nova/1.0", libvirt.DOMAIN_AFFECT_CONFIG)
+		if err != nil {
+			log.Fatalf("Err: %v", err)
+			return err
+		}
+		vmInfo, err := parseNovaMetadata(metadata)
+		if err != nil {
+			log.Fatalf("Err: %v", err)
+			return err
+		}
+		vmInfoJSON, err := json.Marshal(vmInfo)
+		if err != nil {
+			log.Fatalf("Err: %v", err)
+			return err
+		}
+		infoPool.Set(fmt.Sprintf("vm/%s/%s", name, "vminfo"), string(vmInfoJSON))
+
+		// Get Network info
+		xml, err := d.GetXMLDesc(0)
+		if err != nil {
+			log.Fatalf("Err: %v", err)
+			return err
+		}
+		ifInfo, err := parseXMLForMAC(xml)
+		if err != nil {
+			log.Fatalf("Err: %v", err)
+			return err
+		}
+		setInterfaceAnnotation(ifInfo, vmIfInfoChan)
+
+		ifInfoJSON, err := json.Marshal(ifInfo)
+		if err != nil {
+			log.Fatalf("Err: %v", err)
+			return err
+		}
+		infoPool.Set(fmt.Sprintf("vm/%s/%s", name, "interfaces"), string(ifInfoJSON))
+	}
+	return nil
+}
+
+// RunVirshEventLoop is event loop to watch libvirt update
+func RunVirshEventLoop(ctx context.Context, conn *libvirt.Connect, vmIfInfoChan chan string) error {
+	callbackID, err := conn.DomainEventLifecycleRegister(nil, domainEventLifecycleCallback(vmIfInfoChan))
+	if err != nil {
+		log.Fatalf("Err: callbackid: %d %v", callbackID, err)
+	}
+
+	libvirt.EventAddTimeout(5000, func(timer int) { return }) // 5000 = 5sec
+	log.Printf("Entering libvirt event loop()")
+EVENTLOOP:
+	for {
+		select {
+		case <-ctx.Done():
+			break EVENTLOOP
+		default:
+			if err := libvirt.EventRunDefaultImpl(); err != nil {
+				log.Fatalf("%v", err)
+			}
+		}
+	}
+	log.Printf("Quitting libvirt event loop()")
+
+	if err := conn.DomainEventDeregister(callbackID); err != nil {
+		log.Fatalf("%v", err)
+	}
+	return nil
+}
diff --git a/src/dma/cmd/server/agent.go b/src/dma/cmd/server/agent.go
new file mode 100644
index 00000000..ffcb4a97
--- /dev/null
+++ b/src/dma/cmd/server/agent.go
@@ -0,0 +1,31 @@
+package main
+
+import (
+	"fmt"
+	"os/exec"
+	"strings"
+)
+
+func createCollectdConf() error {
+	outStatus, errStatus := exec.Command("ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", "localhost", "sudo", "systemctl", "status", "collectd").Output()
+	if errStatus != nil {
+		return fmt.Errorf("status NG")
+	}
+	if !strings.Contains(string(outStatus), "running") {
+		return fmt.Errorf("status not running")
+	}
+
+	_, errStop := exec.Command("ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", "localhost", "sudo", "systemctl", "stop", "collectd").Output()
+	if errStop != nil {
+		return fmt.Errorf("stop NG")
+	}
+
+	_, errStart := exec.Command("ssh", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", "localhost", "sudo", "systemctl", "start", "collectd").Output()
+	if errStart != nil {
+		return fmt.Errorf("start NG")
+	}
+
+	fmt.Println("All complete!")
+
+	return nil
+}
diff --git a/src/dma/cmd/server/amqp.go b/src/dma/cmd/server/amqp.go
new file mode 100644
index 00000000..4d080fe1
--- /dev/null
+++ b/src/dma/cmd/server/amqp.go
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2017 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"github.com/streadway/amqp"
+	"log"
+	"os"
+	"strings"
+)
+
+func failOnError(err error, msg string) {
+	if err != nil {
+		log.Fatalf("%s: %s", msg, err)
+	}
+}
+
+func runSubscriber(ctx context.Context, config *Config) {
+	confDirPath := config.Server.CollectdConfDir
+	amqpURL := "amqp://" + config.Server.AmqpUser + ":" + config.Server.AmqpPassword + "@" + config.Server.AmqpHost + ":" + config.Server.AmqpPort + "/"
+	conn, err := amqp.Dial(amqpURL)
+	failOnError(err, "Failed to connect to RabbitMQ")
+
+	defer conn.Close()
+
+	ch, err := conn.Channel()
+	failOnError(err, "Failed to open a channel")
+	defer ch.Close()
+
+	err = ch.ExchangeDeclare(
+		"collectd-conf", // name
+		"fanout",        // type
+		false,           // durable
+		false,           // auto-deleted
+		false,           // internal
+		false,           // no-wait
+		nil,             // arguments
+	)
+	failOnError(err, "Failed to declare an exchange")
+
+	q, err := ch.QueueDeclare(
+		"",    // name
+		false, // durable
+		false, // delete when unused
+		true,  // exclusive
+		false, // no-wait
+		nil,   // arguments
+	)
+	failOnError(err, "Failed to declare a queue")
+
+	err = ch.QueueBind(
+		q.Name,          // queue name
+		"",              // routing key
+		"collectd-conf", // exchange
+		false,
+		nil)
+	failOnError(err, "Failed to bind a queue")
+
+	msgs, err := ch.Consume(
+		q.Name, // queue
+		"",     // consumer
+		true,   // auto-ack
+		false,  // exclusive
+		false,  // no-local
+		false,  // no-wait
+		nil,    // args
+	)
+	failOnError(err, "Failed to register a consumer")
+
+EVENTLOOP:
+	for {
+		select {
+		case <-ctx.Done():
+			break EVENTLOOP
+		case d, ok := <-msgs:
+			if ok {
+				dataText := strings.SplitN(string(d.Body), "/", 2)
+
+				dst, err := os.Create(confDirPath + "/" + dataText[0])
+				failOnError(err, "File create NG")
+				defer dst.Close()
+
+				dst.Write(([]byte)(dataText[1]))
+
+				err = createCollectdConf()
+				failOnError(err, "collectd conf NG")
+
+				log.Printf(" [x] %s", d.Body)
+			}
+		}
+	}
+
+}
diff --git a/src/dma/cmd/server/api.go b/src/dma/cmd/server/api.go
new file mode 100644
index 00000000..e8add0a1
--- /dev/null
+++ b/src/dma/cmd/server/api.go
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2017 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/labstack/echo"
+	"io"
+	"net/http"
+	"os"
+	"time"
+)
+
+func runAPIServer(ctx context.Context, config *Config) {
+	confDirPath := config.Server.CollectdConfDir
+	e := echo.New()
+
+	e.GET("/", func(c echo.Context) error {
+		return c.String(http.StatusOK, "GET OK")
+	})
+	e.POST("/collectd/conf", func(c echo.Context) error {
+
+		file, err := c.FormFile("file")
+		if err != nil {
+			return c.String(http.StatusInternalServerError, "file send NG")
+		}
+
+		src, err := file.Open()
+		if err != nil {
+			return c.String(http.StatusInternalServerError, "file open NG")
+		}
+		defer src.Close()
+
+		dst, err := os.Create(confDirPath + "/" + file.Filename)
+		if err != nil {
+			return c.String(http.StatusInternalServerError, "file create NG")
+		}
+		defer dst.Close()
+
+		// Copy
+		if _, err = io.Copy(dst, src); err != nil {
+			return c.String(http.StatusInternalServerError, "file write NG")
+		}
+
+		err = createCollectdConf()
+		if err != nil {
+			errstr := fmt.Sprintf("collectd conf NG:%v", err)
+			return c.String(http.StatusInternalServerError, errstr)
+		}
+		return c.String(http.StatusCreated, "collectd conf OK")
+	})
+
+	// Start server
+	go func() {
+		urlStr := ":" + config.Server.ListenPort
+		if err := e.Start(urlStr); err != nil {
+			e.Logger.Info("shutting down the server")
+		}
+	}()
+
+	// Wait for context.Done() to gracefully shutdown the server with
+	// a timeout of 10 seconds.
+	<-ctx.Done()
+	ctxShutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	if err := e.Shutdown(ctxShutdown); err != nil {
+		e.Logger.Fatal(err)
+	}
+
+}
diff --git a/src/dma/cmd/server/main.go b/src/dma/cmd/server/main.go
new file mode 100644
index 00000000..2e028fa4
--- /dev/null
+++ b/src/dma/cmd/server/main.go
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2017 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"flag"
+	"github.com/BurntSushi/toml"
+	"log"
+	"os"
+	"sync"
+)
+
+var serverTypeOpt = flag.String("type", "both", "server type: both(default), pubsub, rest")
+
+// Config is ...
+type Config struct {
+	Server ServerConfig
+}
+
+// ServerConfig is ...
+type ServerConfig struct {
+	ListenPort string `toml:"listen_port"`
+
+	AmqpHost     string `toml:"amqp_host"`
+	AmqpUser     string `toml:"amqp_user"`
+	AmqpPassword string `toml:"amqp_password"`
+	AmqpPort     string `toml:"amqp_port"`
+
+	CollectdConfDir string `toml:"collectd_confdir"`
+}
+
+func main() {
+
+	var config Config
+	_, err := toml.DecodeFile("/etc/barometer-dma/config.toml", &config)
+	if err != nil {
+		log.Fatalf("Read error of config file")
+	}
+
+	if f, err := os.Stat(config.Server.CollectdConfDir); os.IsNotExist(err) || !f.IsDir() {
+		log.Fatalf("Path \"%s\" is not a directory", config.Server.CollectdConfDir)
+	}
+
+	var waitgroup sync.WaitGroup
+
+	flag.Parse()
+
+	if *serverTypeOpt == "both" || *serverTypeOpt == "pubsub" {
+		ctx := context.Background()
+		waitgroup.Add(1)
+		go func() {
+			defer waitgroup.Done()
+			runSubscriber(ctx, &config)
+		}()
+		log.Printf("Waiting for publish.")
+	}
+
+	if *serverTypeOpt == "both" || *serverTypeOpt == "rest" {
+		ctx := context.Background()
+		waitgroup.Add(1)
+		go func() {
+			defer waitgroup.Done()
+			runAPIServer(ctx, &config)
+		}()
+		log.Printf("Waiting for REST.")
+	}
+
+	waitgroup.Wait()
+	log.Printf("Server stop.")
+}
diff --git a/src/dma/cmd/threshold/evaluate.go b/src/dma/cmd/threshold/evaluate.go
new file mode 100644
index 00000000..8c961253
--- /dev/null
+++ b/src/dma/cmd/threshold/evaluate.go
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2018 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+func evaluate(config *Config, rdlist []rawData) []evalData {
+	edlist := []evalData{}
+
+	for _, rd := range rdlist {
+		maxVal := 0
+		for _, val := range rd.datalist {
+			if maxVal < val {
+				maxVal = val
+			}
+		}
+
+		if maxVal > config.Threshold.Min {
+			edlist = append(edlist, evalData{rd.key, 1})
+		} else {
+			edlist = append(edlist, evalData{rd.key, 0})
+		}
+	}
+	return edlist
+}
diff --git a/src/dma/cmd/threshold/main.go b/src/dma/cmd/threshold/main.go
new file mode 100644
index 00000000..b98328d1
--- /dev/null
+++ b/src/dma/cmd/threshold/main.go
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2018 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/BurntSushi/toml"
+	"log"
+	"sync"
+	"time"
+)
+
+// Config is ...
+type Config struct {
+	Common    CommonConfig
+	Threshold ThresholdConfig
+}
+
+// CommonConfig is ...
+type CommonConfig struct {
+	RedisHost     string `toml:"redis_host"`
+	RedisPort     string `toml:"redis_port"`
+	RedisPassword string `toml:"redis_password"`
+	RedisDB       int    `toml:"redis_db"`
+}
+
+// ThresholdConfig is ...
+type ThresholdConfig struct {
+	RedisHost     string `toml:"redis_host"`
+	RedisPort     string `toml:"redis_port"`
+	RedisPassword string `toml:"redis_password"`
+	RedisDB       int    `toml:"redis_db"`
+
+	Interval int `toml:"interval"`
+	Min      int `toml:"min"`
+
+	CollectdPlugin string `toml:"collectd_plugin"`
+	CollectdType   string `toml:"collectd_type"`
+}
+
+type rawData struct {
+	key      string
+	datalist []int
+}
+
+type evalData struct {
+	key   string
+	label int
+}
+
+func main() {
+	var config Config
+	_, err := toml.DecodeFile("/etc/barometer-dma/config.toml", &config)
+	if err != nil {
+		log.Fatalf("Read error of config: %s", err)
+	}
+
+	thresConfig := config.Threshold
+	log.Printf("Raw data redis config Addr:%s:%s DB:%d", thresConfig.RedisHost, thresConfig.RedisPort, thresConfig.RedisDB)
+	if thresConfig.RedisPassword == "" {
+		log.Printf("Raw data redis password is not set")
+	}
+	annoConfig := config.Common
+	log.Printf("Annotate redis config Addr:%s:%s DB:%d", annoConfig.RedisHost, annoConfig.RedisPort, annoConfig.RedisDB)
+	if annoConfig.RedisPassword == "" {
+		log.Printf("Annotate redis password is not set")
+	}
+
+	var waitgroup sync.WaitGroup
+	ctx := context.Background()
+
+	waitgroup.Add(1)
+	go func() {
+		defer waitgroup.Done()
+		ticker := time.NewTicker(time.Duration(config.Threshold.Interval) * time.Second)
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				result1 := read(&config)
+				// analysis()
+				result2 := evaluate(&config, result1)
+				transmit(&config, result2)
+			}
+		}
+	}()
+
+	waitgroup.Wait()
+
+	fmt.Println("End")
+}
diff --git a/src/dma/cmd/threshold/read.go b/src/dma/cmd/threshold/read.go
new file mode 100644
index 00000000..0b0cf5a1
--- /dev/null
+++ b/src/dma/cmd/threshold/read.go
@@ -0,0 +1,82 @@
+/*
+ * Copyright 2018 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"fmt"
+	"github.com/go-redis/redis"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// e.g. collectd/instance-00000001/virt/if_octets-tapd21acb51-35
+const redisKey = "collectd/*/virt/if_octets-*"
+
+func zrangebyscore(config *Config, client *redis.Client, key string) []int {
+
+	unixNow := int(time.Now().Unix())
+
+	val, err := client.ZRangeByScore(key, redis.ZRangeBy{
+		Min: strconv.Itoa(unixNow - config.Threshold.Interval),
+		Max: strconv.Itoa(unixNow),
+	}).Result()
+
+	datalist := []int{}
+
+	if err == redis.Nil {
+		fmt.Println("this key is not exist")
+		os.Exit(1)
+	} else if err != nil {
+		panic(err)
+	} else {
+		for _, strVal := range val {
+			split := strings.Split(strVal, ":")
+			txVal := split[2]
+			floatVal, err := strconv.ParseFloat(txVal, 64)
+			if err != nil {
+				os.Exit(1)
+			}
+			datalist = append(datalist, int(floatVal))
+		}
+	}
+	return datalist
+}
+
+func read(config *Config) []rawData {
+	thresConfig := config.Threshold
+
+	client := redis.NewClient(&redis.Options{
+		Addr:     thresConfig.RedisHost + ":" + thresConfig.RedisPort,
+		Password: thresConfig.RedisPassword,
+		DB:       thresConfig.RedisDB,
+	})
+
+	keys, err := client.Keys(redisKey).Result()
+	if err != nil {
+		panic(err)
+	}
+
+	rdlist := []rawData{}
+
+	for _, key := range keys {
+		rdlist = append(rdlist, rawData{key, zrangebyscore(config, client, key)})
+	}
+
+	return rdlist
+}
diff --git a/src/dma/cmd/threshold/transmit.go b/src/dma/cmd/threshold/transmit.go
new file mode 100644
index 00000000..8cac2a88
--- /dev/null
+++ b/src/dma/cmd/threshold/transmit.go
@@ -0,0 +1,97 @@
+/*
+ * Copyright 2018 NEC Corporation
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"github.com/distributed-monitoring/agent/pkg/common"
+	"github.com/go-redis/redis"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type collectdNotifier struct {
+	pluginName string
+	typeName   string
+}
+
+func send(cn collectdNotifier, message string, severity string, metaData [][2]string) error {
+	unixNow := float64(time.Now().UnixNano()) / 1000000000
+
+	var metaDataStr bytes.Buffer
+	for _, data := range metaData {
+		metaDataStr.WriteString(" s:")
+		metaDataStr.WriteString(data[0])
+		metaDataStr.WriteString("=\"")
+		metaDataStr.WriteString(strings.Replace(data[1], "\"", "\\\"", -1))
+		metaDataStr.WriteString("\"")
+	}
+
+	fmt.Printf("PUTNOTIF message=\"%s\" severity=%s time=%f "+
+		"host=localhost plugin=%s type=%s %s\n",
+		message, severity, unixNow, cn.pluginName, cn.typeName, metaDataStr.String())
+
+	return nil
+}
+
+func transmit(config *Config, edlist []evalData) {
+	annoConfig := config.Common
+
+	client := redis.NewClient(&redis.Options{
+		Addr:     annoConfig.RedisHost + ":" + annoConfig.RedisPort,
+		Password: annoConfig.RedisPassword,
+		DB:       annoConfig.RedisDB,
+	})
+	pool := common.RedisPool{Client: client}
+
+	notifier := collectdNotifier{
+		pluginName: config.Threshold.CollectdPlugin,
+		typeName:   config.Threshold.CollectdType}
+
+	for _, ed := range edlist {
+		if ed.label == 1 {
+
+			fmt.Println("kick action")
+
+			item := strings.Split(ed.key, "/")
+			ifItem := strings.SplitN(item[3], "-", 2)
+			virtName := item[1]
+			virtIF := ifItem[1]
+
+			var message bytes.Buffer
+			message.WriteString("Value exceeded threshold ")
+			message.WriteString(strconv.Itoa(config.Threshold.Min))
+			message.WriteString(".")
+
+			nameVal, _ := pool.Get(fmt.Sprintf("%s/%s/vminfo", "vm", virtName))
+			ifVal, _ := pool.Get(fmt.Sprintf("%s/%s/neutron_network", "if", virtIF))
+
+			nameInfo := fmt.Sprintf("{\"%s\": %s}", virtName, nameVal)
+			ifInfo := fmt.Sprintf("{\"%s\": %s}", virtIF, ifVal)
+
+			fmt.Println(nameInfo)
+			fmt.Println(ifInfo)
+
+			send(notifier, message.String(),
+				"warning",
+				[][2]string{{"vminfo", nameInfo}, {"neutron_network", ifInfo}})
+
+		}
+	}
+}