<commit_before>package main

/* TODO:

1. [x] serve HTTP /v1/instances/:app_id to retrieve ip:port pairs for given app
2. [x] implement UDP proxy (one-way/two-way, fanout (& roundrobin))
3. [x] implement upstream-conf.d file management
4. [ ] implement TCP proxy (with pluggable impls: haproxy, lvs, ...)
5. [ ] tcp-proxy: add `accept-proxy` support
6. [ ] tcp-proxy: add `proxy-protocol` support
7. [ ] logging: add readable up/down notices, such as:
	- "APP_ID: Task $TASK_STATUS on host $HOSTNAME ($TASK_ID)."
		with TASK_STATUS translating into readable ("killed", "finished", "running", ...)
	- "APP_ID: Task on $HOSTNAME:$PORT is unhealthy ($TASK_ID)."
	- "APP_ID: Task on $HOSTNAME:$PORT is healthy ($TASK_ID)."
8. [ ] implement HTTP(S) gateway support

XXX Changes:

* `--marathon-host` is now named `--marathon-ip` and only accepts IP addresses
* `--reconnect-delay` added
* `--haproxy-cfg` removed
* `--gateway` renamed to `--enable-gateway`
* `--enable-files` added
* `--enable-tcp` added
* `--enable-udp` added
* also exposes service discovery API via HTTP endpoint

*/

import (
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/christianparpart/serviced/marathon"
	"github.com/gorilla/mux"
	flag "github.com/ogier/pflag"
)

type mmsdHandler interface {
	Apply(apps []*marathon.App, force bool) error
	Update(app *marathon.App, task *marathon.Task) error
	IsEnabled() bool
	SetEnabled(value bool)
}

type mmsdService struct {
	MarathonScheme   string
	MarathonIP       net.IP
	MarathonPort     uint
	ReconnectDelay   time.Duration
	RunStateDir      string
	FilterGroups     string
	GatewayEnabled   bool
	GatewayPortHTTP  uint
	GatewayPortHTTPS uint
	ManagedIP        net.IP
	FilesEnabled     bool
	UDPEnabled       bool
	TCPEnabled       bool
	HaproxyBin       string
	HaproxyTailCfg   string
	HaproxyPort      uint
	ServiceBind      net.IP
	ServicePort      uint
	Verbose          bool
	Handlers         []mmsdHandler
	quitChannel      chan bool
}

func (mmsd *mmsdService) setupHttpService() {
	router := mux.NewRouter()
	router.HandleFunc("/ping", mmsd.v0Ping)
	router.HandleFunc("/version", mmsd.v0Version)

	v1 := router.PathPrefix("/v1").Subrouter()
	v1.HandleFunc("/apps", mmsd.v1Apps).Methods("GET")
	v1.HandleFunc("/instances{app_id:/.*}", mmsd.v1Instances).Methods("GET")

	serviceAddr := fmt.Sprintf("%v:%v", mmsd.ServiceBind, mmsd.ServicePort)
	log.Printf("Exposing service API on http://%v\n", serviceAddr)

	go http.ListenAndServe(serviceAddr, router)
}

func (mmsd *mmsdService) v0Ping(w http.ResponseWriter, r *http.Request) {
	fmt.Fprint(w, "pong\n")
}

func (mmsd *mmsdService) v0Version(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintf(w, "mmsd %v\n", appVersion)
}

func (mmsd *mmsdService) v1Apps(w http.ResponseWriter, r *http.Request) {
	m, err := marathon.NewService(mmsd.MarathonIP, mmsd.MarathonPort)

	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("NewService error. %v\n", err)
		return
	}

	apps, err := m.GetApps()
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("GetApps error. %v\n", err)
		return
	}

	var appList []string
	for _, app := range apps {
		appList = append(appList, app.Id)
	}
	sort.Strings(appList)

	fmt.Fprintf(w, "%s\n", strings.Join(appList, "\n"))
}

func (mmsd *mmsdService) v1Instances(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	appID := vars["app_id"]
	noResolve := r.URL.Query().Get("noresolve") == "1"

	var portIndex int
	if sval := r.URL.Query().Get("portIndex"); len(sval) != 0 {
		i, err := strconv.Atoi(sval)
		if err == nil {
			portIndex = i
		}
	}

	m, err := marathon.NewService(mmsd.MarathonIP, mmsd.MarathonPort)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("NewService error. %v\n", err)
		return
	}

	app, err := m.GetApp(appID)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("GetApp error. %v\n", err)
		return
	}

	if app == nil {
		w.WriteHeader(http.StatusNotFound)
		return
	}

	if len(app.Tasks) == 0 {
		w.WriteHeader(http.StatusNoContent)
		return
	}

	// appJson, err := json.MarshalIndent(app, "", " ")
	// w.Write(appJson)<commit_msg>http-api: return 400 bad request when portIndex is out of bounds<commit_after>package main

/* TODO:

1. [x] serve HTTP /v1/instances/:app_id to retrieve ip:port pairs for given app
2. [x] implement UDP proxy (one-way/two-way, fanout (& roundrobin))
3. [x] implement upstream-conf.d file management
4. [ ] implement TCP proxy (with pluggable impls: haproxy, lvs, ...)
5. [ ] tcp-proxy: add `accept-proxy` support
6. [ ] tcp-proxy: add `proxy-protocol` support
7. [ ] logging: add readable up/down notices, such as:
	- "APP_ID: Task $TASK_STATUS on host $HOSTNAME ($TASK_ID)."
		with TASK_STATUS translating into readable ("killed", "finished", "running", ...)
	- "APP_ID: Task on $HOSTNAME:$PORT is unhealthy ($TASK_ID)."
	- "APP_ID: Task on $HOSTNAME:$PORT is healthy ($TASK_ID)."
8. [ ] implement HTTP(S) gateway support

XXX Changes:

* `--marathon-host` is now named `--marathon-ip` and only accepts IP addresses
* `--reconnect-delay` added
* `--haproxy-cfg` removed
* `--gateway` renamed to `--enable-gateway`
* `--enable-files` added
* `--enable-tcp` added
* `--enable-udp` added
* also exposes service discovery API via HTTP endpoint

*/

import (
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/christianparpart/serviced/marathon"
	"github.com/gorilla/mux"
	flag "github.com/ogier/pflag"
)

type mmsdHandler interface {
	Apply(apps []*marathon.App, force bool) error
	Update(app *marathon.App, task *marathon.Task) error
	IsEnabled() bool
	SetEnabled(value bool)
}

type mmsdService struct {
	MarathonScheme   string
	MarathonIP       net.IP
	MarathonPort     uint
	ReconnectDelay   time.Duration
	RunStateDir      string
	FilterGroups     string
	GatewayEnabled   bool
	GatewayPortHTTP  uint
	GatewayPortHTTPS uint
	ManagedIP        net.IP
	FilesEnabled     bool
	UDPEnabled       bool
	TCPEnabled       bool
	HaproxyBin       string
	HaproxyTailCfg   string
	HaproxyPort      uint
	ServiceBind      net.IP
	ServicePort      uint
	Verbose          bool
	Handlers         []mmsdHandler
	quitChannel      chan bool
}

func (mmsd *mmsdService) setupHttpService() {
	router := mux.NewRouter()
	router.HandleFunc("/ping", mmsd.v0Ping)
	router.HandleFunc("/version", mmsd.v0Version)

	v1 := router.PathPrefix("/v1").Subrouter()
	v1.HandleFunc("/apps", mmsd.v1Apps).Methods("GET")
	v1.HandleFunc("/instances{app_id:/.*}", mmsd.v1Instances).Methods("GET")

	serviceAddr := fmt.Sprintf("%v:%v", mmsd.ServiceBind, mmsd.ServicePort)
	log.Printf("Exposing service API on http://%v\n", serviceAddr)

	go http.ListenAndServe(serviceAddr, router)
}

func (mmsd *mmsdService) v0Ping(w http.ResponseWriter, r *http.Request) {
	fmt.Fprint(w, "pong\n")
}

func (mmsd *mmsdService) v0Version(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintf(w, "mmsd %v\n", appVersion)
}

func (mmsd *mmsdService) v1Apps(w http.ResponseWriter, r *http.Request) {
	m, err := marathon.NewService(mmsd.MarathonIP, mmsd.MarathonPort)

	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("NewService error. %v\n", err)
		return
	}

	apps, err := m.GetApps()
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("GetApps error. %v\n", err)
		return
	}

	var appList []string
	for _, app := range apps {
		appList = append(appList, app.Id)
	}
	sort.Strings(appList)

	fmt.Fprintf(w, "%s\n", strings.Join(appList, "\n"))
}

func (mmsd *mmsdService) v1Instances(w http.ResponseWriter, r *http.Request) {
	vars := mux.Vars(r)
	appID := vars["app_id"]
	noResolve := r.URL.Query().Get("noresolve") == "1"

	var portIndex int
	if sval := r.URL.Query().Get("portIndex"); len(sval) != 0 {
		i, err := strconv.Atoi(sval)
		if err == nil {
			portIndex = i
		}
	}

	m, err := marathon.NewService(mmsd.MarathonIP, mmsd.MarathonPort)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("NewService error. %v\n", err)
		return
	}

	app, err := m.GetApp(appID)
	if err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		log.Printf("GetApp error. %v\n", err)
		return
	}

	if app == nil {
		w.WriteHeader(http.StatusNotFound)
		return
	}

	if portIndex >= len(app.Ports) {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	if len(app.Tasks) == 0 {
		w.WriteHeader(http.StatusNoContent)
		return
	}

	// appJson, err := json.MarshalIndent(app, "", " ")
	// w.Write(appJson)
{'fraction_non_alphanumeric': 0.14556596138281883, 'fraction_numerical': 0.005896635449184877, 'mean_word_length': 3.8378076062639823, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2378202', 'n_tokens_mistral': 3360, 'n_tokens_neox': 3252, 'n_words': 750}