func (t *HTTPProvider) Init(i interface{}) error {
	conf := i.(*HTTPConfig)

	// make sure the port is open
	l, err := net.Listen("tcp", conf.Listen)
	if err != nil {
		logrus.Error(err)

		// check to see if the busy port is due to another provider
		resp, err := std_http.Get("http://" + conf.Listen + ENDPOINT + "?init_check=true")
		if err != nil {
			return err
		}

		buff, _ := ioutil.ReadAll(resp.Body)
		if string(buff) != START_HANDSHAKE {
			return err

		}
	}

	// stop the test
	err = l.Close()
	if err != nil {
		logrus.Error(err)
		return err
	}

	// update the providers litening address
	t.listen = conf.Listen
	t.pool = event.NewEncodingPool(event.EncoderFactories[conf.Encoding], event.DecoderFactories[conf.Encoding], conf.MaxEncoders)

	return nil
}

func (v *vm) Machines() ([]vms.Machine, error) {
	resp, err := http.Get("http://" + v.server + ":8080/api/iaas")
	if err != nil {
		log.Error(err)
		return nil, err
	}
	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Error(err)
		return nil, err
	}
	type Status struct {
		Id         string `json:"id"`
		Type       string `json:"type"`
		Attributes VmInfo
	}
	var State struct {
		Data []Status `json:"data"`
	}

	err = json.Unmarshal(body, &State)
	if err != nil {
		log.Error(err)
		return nil, err
	}
	var machines = make([]vms.Machine, len(State.Data))
	for i, val := range State.Data {
		machines[i] = &machine{id: val.Id, server: v.server}
	}
	return machines, nil
}

func (i *Incident) Get(w http.ResponseWriter, r *http.Request) {
	w.Header().Add("content-type", "application/json")
	vars := mux.Vars(r)
	id, ok := vars["id"]
	if !ok {
		http.Error(w, "Must append incident id", http.StatusBadRequest)
		return
	}
	index := i.pipeline.GetIndex()

	// if the id is "*", fetch all outstanding incidents
	if id == "*" {
		all := index.ListIncidents()
		all = reduceStatusAbove(event.WARNING, all)
		buff, err := json.Marshal(makeKV(all))
		if err != nil {
			logrus.Error(err)
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		w.Write(buff)
		return
	}

	// write out the found incident. The value will be null if nothing was found
	in := index.GetIncident([]byte(id))
	buff, err := json.Marshal(in)
	if err != nil {
		logrus.Error(err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.Write(buff)
}

func getTgtFromMountPoint(mountpoint string) (target string, portal string) {
	log.Infof("Get iSCSI target for path %s", mountpoint)
	out, err := exec.Command("sudo", "df", "--output=source", mountpoint).CombinedOutput()
	if err != nil {
		log.Error("Failed to obtain device info from df cmd: ", err)
		return
	}

	device := "../../" + strings.Split(strings.Fields(string(out))[1], "/")[2]
	log.Debug("Formed the device: ", device)
	out, err = exec.Command("sudo", "ls", "-l", "/dev/disk/by-path").CombinedOutput()
	if err != nil {
		log.Error("Failed to list contents of /dev/disk/by-path/: ", err)
		return
	}
	lines := strings.Split(string(out), "\n")
	for _, line := range lines {
		log.Debugf("check line %s for %s...", line, device)
		if strings.Contains(line, device) {
			target = strings.Split((strings.Split(line, "-iscsi-")[1]), "-lun-")[0]
			portal = strings.Split((strings.Split(line, " ip-")[1]), "-iscsi-")[0]
		}
	}
	return
}

// GetComputeSystemProperties gets the properties for the compute system with the given ID.
func GetComputeSystemProperties(id string, flags uint32) (ComputeSystemProperties, error) {
	title := "hcsshim::GetComputeSystemProperties "

	csProps := ComputeSystemProperties{
		Stopped:           false,
		AreUpdatesPending: false,
	}

	logrus.Debugf("Calling proc")
	var buffer *uint16
	err := getComputeSystemProperties(id, flags, &buffer)
	if err != nil {
		err = makeError(err, title, "")
		logrus.Error(err)
		return csProps, err
	}
	propData := convertAndFreeCoTaskMemString(buffer)
	logrus.Debugf(title+" - succeeded output=%s", propData)

	if err = json.Unmarshal([]byte(propData), &csProps); err != nil {
		logrus.Error(err)
		return csProps, err
	}

	return csProps, nil
}

func handlePullRequestReviewComment(w http.ResponseWriter, r *http.Request) {
	hook, err := github.ParsePullRequestReviewCommentHook(r.Body)
	if err != nil {
		logrus.Error(err)
		w.WriteHeader(500)
		return
	}

	if !hook.IsOpen() {
		w.WriteHeader(200)
		return
	}

	g := github.GitHub{
		AuthToken: config.GHToken,
		User:      config.GHUser,
	}

	if err := g.MoveTriageForward(hook.Repo, hook.PullRequest.Number, hook.Comment); err != nil {
		logrus.Error(err)
		w.WriteHeader(500)
		return
	}

	w.WriteHeader(204)
	return
}

func (c *controller) processNetworkUpdate(nws []*store.KVPair, prune *networkTable) {
	for _, kve := range nws {
		var n network
		err := json.Unmarshal(kve.Value, &n)
		if err != nil {
			log.Error(err)
			continue
		}
		if prune != nil {
			delete(*prune, n.id)
		}
		n.SetIndex(kve.LastIndex)
		c.Lock()
		existing, ok := c.networks[n.id]
		c.Unlock()
		if ok {
			existing.Lock()
			// Skip existing network update
			if existing.dbIndex != n.Index() {
				// Can't use SetIndex() since existing is locked.
				existing.dbIndex = n.Index()
				existing.dbExists = true
				existing.endpointCnt = n.endpointCnt
			}
			existing.Unlock()
			continue
		}

		if err = c.newNetworkFromStore(&n); err != nil {
			log.Error(err)
		}
	}
}

func (upd *updater) doUpdate(update model.ServiceUpdate) {
	svc := upd.services[update.ServiceKey]
	if svc == nil {
		if update.ServiceInfo == nil {
			return
		}

		svc, err := upd.config.newService(update, upd.errors)
		if err != nil {
			log.Error("adding service ", update.ServiceKey, ": ",
				err)
			return
		}

		upd.services[update.ServiceKey] = svc
	} else if update.ServiceInfo != nil {
		err := svc.update(update)
		if err != nil {
			log.Error("updating service ", update.ServiceKey, ": ",
				err)
			return
		}
	} else {
		delete(upd.services, update.ServiceKey)
		svc.close()
	}
}

// Takes a JFile, a JSON path (optional) and JSON data.
// Stores the JSON data. Returns true if successful.
func jfileAdd(L *lua.LState) int {
	jfile := checkJFile(L) // arg 1
	top := L.GetTop()
	jsonpath := "x"
	jsondata := ""
	if top == 2 {
		jsondata = L.ToString(2)
		if jsondata == "" {
			L.ArgError(2, "JSON data expected")
		}
	} else if top == 3 {
		jsonpath = L.ToString(2)
		// Check for { to help avoid allowing JSON data as a JSON path
		if jsonpath == "" || strings.Contains(jsonpath, "{") {
			L.ArgError(2, "JSON path expected")
		}
		jsondata = L.ToString(3)
		if jsondata == "" {
			L.ArgError(3, "JSON data expected")
		}
	}
	err := jfile.AddJSON(jsonpath, []byte(jsondata))
	if err != nil {
		if top == 2 || strings.HasPrefix(err.Error(), "invalid character") {
			log.Error("JSON data: ", err)
		} else {
			log.Error(err)
		}
	}
	L.Push(lua.LBool(err == nil))
	return 1 // number of results
}

//StatsListener Listen for requests to serve server stats
func (s *StatsMgr) StatsListener(ip net.IP, port int, ctrlChan chan int) {
	s.wg.Add(1)

	go func() {
		r := mux.NewRouter()
		r.HandleFunc("/api/v1/stats/all", s.StatsAllJSON)
		svr := &http.Server{}
		svr.Handler = r
		tcpaddr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort(ip.String(), strconv.Itoa(port)))
		if err != nil {
			log.Error("Stats Listener ", err)
			return
		}
		l, err := net.ListenTCP("tcp", tcpaddr)
		log.Printf("%s now listening on %s for incoming connections", strings.Join([]string{AppName, "Stats Listener"}, " "), tcpaddr.String())
		if err != nil {
			log.Error("Stats Listener ", err)
			return
		}
		if err := svr.Serve(l); err != nil {
			log.Errorln(err)
		}
		for item := range ctrlChan {
			if item == -1 {
				err = l.Close()
				if err != nil {
					log.Println(err)
				}
				s.wg.Done()
				log.Printf("%s shutting down", strings.Join([]string{AppName, "Stats Listener"}, " "))
				return
			}
		}
	}()
}

func (riakNode *RiakNode) configureAdvanced(cepmdPort int) {

	fetchURI := fmt.Sprintf("%s/api/v1/clusters/%s/advancedConfig", riakNode.taskData.URI, riakNode.taskData.ClusterName)
	resp, err := http.Get(fetchURI)
	if err != nil {
		log.Error("Unable to fetch advanced config: ", err)
	}
	advancedConfig, err := ioutil.ReadAll(resp.Body)
	resp.Body.Close()
	if err != nil {
		log.Error("Unable to fetch advanced config: ", err)
	}

	tmpl, err := template.New("advanced").Parse(string(advancedConfig))

	if err != nil {
		log.Panic(err)
	}

	// Populate template data from the MesosTask
	vars := advancedTemplateData{}
	vars.CEPMDPort = cepmdPort
	file, err := os.OpenFile("root/riak/etc/advanced.config", os.O_TRUNC|os.O_CREATE|os.O_RDWR, 0664)

	defer file.Close()
	if err != nil {
		log.Panic("Unable to open file: ", err)
	}

	err = tmpl.Execute(file, vars)

	if err != nil {
		log.Panic("Got error", err)
	}
}

// GetFinishedTestRuns returns a map of test UUIDS (keys) and the corresponding post-test metric data for those UUIDs (values)
// The metric data is stored as a string containing JSON text, so this is what's placed into this map (meaning JSON parsing is
// not performed in this function)
func (ac AgentCache) GetFinishedTestRuns() (map[string]string, error) {

	retmap := make(map[string]string)

	// Open connection
	db, err := sql.Open("sqlite3", ac.db_loc)
	if err != nil {
		log.Error(err)
		return retmap, errors.New("Error accessing sqlite cache for finished testruns")
	}
	defer db.Close()

	rows, err := db.Query(fmt.Sprint("select uuid, results from testruns where results != \"\" "))
	if err != nil {
		log.Error(err)
		return retmap, errors.New("Error creating query for selecting finished testruns")
	}

	defer rows.Close()
	for rows.Next() {
		var uuid, testdata string
		rows.Scan(&uuid, &testdata)
		log.Debug("Found ripe testrun: ", uuid)
		retmap[uuid] = testdata
	}

	return retmap, nil
}

// UpdateTestRunData will update an existing testrun entry in the agent cache with the post-test
// metrics dataset that corresponds to that testrun (by testrun UUID)
func (ac AgentCache) UpdateTestRunData(uuid string, testData string) error {

	// Open connection
	db, err := sql.Open("sqlite3", ac.db_loc)
	if err != nil {
		log.Error(err)
		return errors.New("Error accessing sqlite cache for testrun update")
	}
	defer db.Close()

	// Begin Update
	tx, err := db.Begin()
	if err != nil {
		log.Error(err)
		return errors.New("Error beginning new UpdateTestRunData action")
	}

	stmt, err := tx.Prepare(fmt.Sprintf("update testruns set results = '%s' where uuid = '%s' ", testData, uuid))
	if err != nil {
		log.Error(err)
		return errors.New("Error preparing new UpdateTestRunData action")
	}
	defer stmt.Close()
	_, err = stmt.Exec()
	if err != nil {
		log.Error(err)
		return errors.New("Error executing new UpdateTestRunData action")
	}
	tx.Commit()

	log.Infof("Inserted test data for %s into cache", uuid)

	return nil
}

// start accepting connections and consume each of them as they come in
func (h *HTTPProvider) Start(p event.Passer) {
	defer func() {
		if r := recover(); r != nil {
			fmt.Println("Recovered in f", r)
		}
	}()
	std_http.HandleFunc(ENDPOINT, func(w std_http.ResponseWriter, r *std_http.Request) {

		// handle the case where a provider is restarting and needs to check if a listener is a bangarang provider or not
		if r.URL.Query().Get("init_check") == "true" {
			w.Write([]byte(START_HANDSHAKE))
		}
		buff, err := ioutil.ReadAll(r.Body)
		if err != nil {
			logrus.Error(err)
			std_http.Error(w, err.Error(), std_http.StatusInternalServerError)
			return
		}

		e := &event.Event{}
		err = h.pool.Decode(buff, e)

		if err != nil {
			logrus.Error(err)
			std_http.Error(w, err.Error(), std_http.StatusInternalServerError)
			return
		}
		p.Pass(e)
		logrus.Debug("Done processing http event")
	})

	logrus.Infof("Serving http listener on %s", h.listen)
	logrus.Fatal(std_http.ListenAndServe(h.listen, nil))
}

// Get HTTP get method
func (p *EscalationConfig) Get(req *Request) {
	var conf *config.AppConfig
	p.pipeline.ViewConfig(func(cf *config.AppConfig) {
		conf = cf
	})
	vars := mux.Vars(req.r)
	id, ok := vars["id"]
	if !ok {
		logrus.Error("Must append escalation id", req.r.URL.String())
		http.Error(req.w, "must append escalation id", http.StatusBadRequest)
		return
	}

	if id == "*" {
		buff, err := json.Marshal(&conf.Escalations)
		if err != nil {
			logrus.Error(err)
			http.Error(req.w, err.Error(), http.StatusBadRequest)
			return
		}
		req.w.Write(buff)
		return
	}

	coll := conf.Escalations.Collection()
	logrus.Error(coll)
}

func NewCollector(grpcCh chan *GrpcRequest, url, dbName string, interval uint64) (*Collector, error) {
	c, err := client.NewHTTPClient(client.HTTPConfig{
		Addr: url,
	})
	if err != nil {
		return nil, err
	}

	_, _, err = c.Ping(0)
	if err != nil {
		log.Error("can not connect to InfluxDB")
		return nil, err
	}

	q := client.NewQuery("CREATE DATABASE "+dbName, "", "")
	if response, err := c.Query(q); err != nil || response.Error() != nil {
		log.Error("can not create database " + dbName)
		return nil, err
	}

	collector := &Collector{
		grpcCh:   grpcCh,
		url:      url,
		dbName:   dbName,
		interval: interval,
		ch:       make(chan watcherEvent, 16),
		client:   c,
	}
	go collector.loop()
	return collector, nil
}

// PrepareLayer finds a mounted read-write layer matching layerId and enables the
// the filesystem filter for use on that layer.  This requires the paths to all
// parent layers, and is necessary in order to view or interact with the layer
// as an actual filesystem (reading and writing files, creating directories, etc).
// Disabling the filter must be done via UnprepareLayer.
func PrepareLayer(info DriverInfo, layerId string, parentLayerPaths []string) error {
	title := "hcsshim::PrepareLayer "
	logrus.Debugf(title+"flavour %d layerId %s", info.Flavour, layerId)

	// Generate layer descriptors
	layers, err := layerPathsToDescriptors(parentLayerPaths)
	if err != nil {
		return err
	}

	// Convert info to API calling convention
	infop, err := convertDriverInfo(info)
	if err != nil {
		logrus.Error(err)
		return err
	}

	err = prepareLayer(&infop, layerId, layers)
	if err != nil {
		err = makeErrorf(err, title, "layerId=%s flavour=%d", layerId, info.Flavour)
		logrus.Error(err)
		return err
	}

	logrus.Debugf(title+"succeeded flavour=%d layerId=%s", info.Flavour, layerId)
	return nil
}

func HandleReadConnection(conn net.Conn, readQueue *lane.Queue, writePort int, mapOperation SplitterMap) {
	buffConn := bufio.NewReaderSize(conn, 1024)
	buffer := make([]byte, 1024)
	for {
		logrus.Debug("Begining Read")
		bytes, err := buffConn.Read(buffer)

		// Output the content of the bytes to the queue
		if bytes > 0 {
			readQueue.Enqueue(buffer[:bytes])
			bytesSeen += bytes
		}

		if bytes == 0 {
			if err.Error() == "EOF" {
				logrus.Error("End of individual transmission")
				buffConn = nil
				conn.Close()
				return
			}
		}

		if err != nil {
			logrus.Error("Underlying network failure?")
			logrus.Error(err)
			break
		}
	}
}

// ListStoppedContainers lists the stopped containers for the specified services.
func (p *Project) ListStoppedContainers(services ...string) ([]string, error) {
	stoppedContainers := []string{}
	err := p.forEach(services, wrapperAction(func(wrapper *serviceWrapper, wrappers map[string]*serviceWrapper) {
		wrapper.Do(nil, EventServiceDeleteStart, EventServiceDelete, func(service Service) error {
			containers, innerErr := service.Containers()
			if innerErr != nil {
				return innerErr
			}

			for _, container := range containers {
				running, innerErr := container.IsRunning()
				if innerErr != nil {
					log.Error(innerErr)
				}
				if !running {
					containerID, innerErr := container.ID()
					if innerErr != nil {
						log.Error(innerErr)
					}
					stoppedContainers = append(stoppedContainers, containerID)
				}
			}

			return nil
		})
	}), nil)
	if err != nil {
		return nil, err
	}
	return stoppedContainers, nil
}

func StartReadListening(readPort int, writePort int, defaultWriters []string, mapOperation SplitterMap) {
	// Create buffer to hold data
	queue := lane.NewQueue()
	MonitorServer(queue)
	// Start listening for writer destinations
	go StartWriteListening(queue, defaultWriters, writePort)

	socket, err := net.Listen("tcp", ":"+strconv.Itoa(readPort))
	if err != nil {
		logrus.Error(err)
	}
	// This will block the main thread
	for {
		// Begin trying to accept connections
		logrus.Debug("Awaiting Connection...")
		//Block and wait for listeners
		conn, err := socket.Accept()
		if err != nil {
			logrus.Error(err)
		} else {
			logrus.Debug("Accepted Connection...")
			go HandleReadConnection(conn, queue, writePort, mapOperation)
		}
	}
}