main.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"time"

	"gopkg.in/alecthomas/kingpin.v2"
)

type (
	rabbitMQConfig struct {
		Exchanges []string `json:"exchanges"`
		Vhost     string   `json:"vhost"`
	}

	fetchedConfig struct {
		RabbitMQConfig rabbitMQConfig `json:"rabbitmq_config"`
		Site           string         `json:"site"`
	}

	config struct {
		Host     string `json:"host"`
		Username string `json:"username"`
		Password string `json:"password"`

		// Services provided at the site of this client
		// this is for  deployment per _service_
		Services []service `json:"services"`

		// GroupToServices maps a group name to services provided for this group
		// this is for  deployment per _group_
		GroupToServices map[string]([]service) `json:"group_to_services"`

		FetchIntervalString    string `json:"fetch_interval"`    // string parsed by time.ParseDuration
		ReconnectTimeoutString string `json:"reconnect_timeout"` // string parsed by time.ParseDuration
		NewTasks               chan task
		DoneTasks              chan taskExecution
		FetchInterval          time.Duration
		ReconnectTimeout       time.Duration
		RabbitMQConfig         rabbitMQConfig
		Site                   string
	}

	// strippedConfig is sent to the backend on startup
	strippedConfig struct {
		Services        []service              `json:"services"`
		GroupToServices map[string]([]service) `json:"group_to_services"`
	}
)

const (
	defaultFetchInterval    = 30 * time.Minute
	defaultReconnectTimeout = 10 * time.Second
)

var (
	client = &http.Client{}
	app    = kingpin.New(
		"FEUDAL Client",
		"Client for the Federated User Credential Deployment Portal (FEUDAL)",
	).Author(
		"Lukas Burgey",
	).Version(
		"0.3.0",
	)
	configFile          = app.Arg("config", "Config file to file to use.").Required().String()
	scriptDebugging     = app.Flag("debug-scripts", "Display debugging info concerning executed scripts").Bool()
	backendDebugging    = app.Flag("debug-backend", "Display debugging info concerning the backend").Bool()
	brokerDebugging     = app.Flag("debug-broker", "Display debugging info concerning the message broker").Bool()
	debugAll            = app.Flag("debug", "Display all debugging info").Bool()
	sequentialExecution = app.Flag("seq", "Execute tasks sequentially").Bool()
)

func logError(err error, msg string) {
	if err != nil {
		log.Printf("[E] %s: %s", msg, err)
	}
}

func (c *config) syncConfig() (err error) {

	log.Printf("[Conf] Synchronising configuration with %v", c.Host)

	var (
		strippedConfigBytes []byte
		req                 *http.Request
		resp                *http.Response
		fetchedConfig       fetchedConfig
	)

	// we inform the backend which services we provide
	strippedConfigBytes, err = json.Marshal(strippedConfig{
		Services:        c.Services,
		GroupToServices: c.GroupToServices,
	})
	if err != nil {
		return
	}

	// update the services tracked by the backend
	req, err = http.NewRequest(
		"PUT",
		"https://"+c.Host+"/backend/clientapi/config",
		bytes.NewReader(strippedConfigBytes),
	)
	if err != nil {
		return
	}

	req.SetBasicAuth(c.Username, c.Password)
	req.Header.Set("Content-Type", "application/json")
	resp, err = client.Do(req)
	if err != nil {
		return
	}
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	if resp.StatusCode != 200 {
		err = fmt.Errorf("Unable to sync configuration (response: %v)", resp.Status)
		return
	}

	err = json.Unmarshal(body, &fetchedConfig)
	if err != nil {
		err = fmt.Errorf("Unable to parse remote configuration: %s %s", err, body)
		return
	}

	c.RabbitMQConfig = fetchedConfig.RabbitMQConfig
	c.Site = fetchedConfig.Site

	log.Printf("[Conf] Synchronised configuration with %v", c.Host)

	return
}

func getConfig(configFile string) (c config, err error) {

	c.Log("Reading config file %s", configFile)

	bs, err := ioutil.ReadFile(configFile)
	if err != nil {
		c.Log("Error reading config file: %s", err)
		return
	}

	err = json.Unmarshal(bs, &c)
	if err != nil {
		c.Log("Error parsing config file: %s", err)
		return
	}

	// check the config values
	if c.Host == "" {
		log.Fatalf("[Conf] No 'host' in config")
	}
	if c.Username == "" {
		log.Fatalf("[Conf] No 'user' in config")
	}
	if c.Password == "" {
		log.Fatalf("[Conf] No 'password' in config")
	}
	if c.FetchIntervalString == "" {
		c.FetchInterval = defaultFetchInterval
		log.Printf("[Conf] Using default fetch_interval of %v", c.FetchInterval)
	} else {
		c.FetchInterval, err = time.ParseDuration(c.FetchIntervalString)
		if err != nil {
			log.Printf("[Conf] Error parsing fetch interval: %s", err)
			err = nil

			c.FetchInterval = defaultFetchInterval
			log.Printf("[Conf] Using default fetch_interval of %v", c.FetchInterval)
		}
	}

	if c.ReconnectTimeoutString == "" {
		c.ReconnectTimeout = defaultReconnectTimeout
		log.Printf("[Conf] Using default reconnect_timeout of %v", c.ReconnectTimeout)
	} else {
		c.ReconnectTimeout, err = time.ParseDuration(c.ReconnectTimeoutString)
		if err != nil {
			log.Printf("[Conf] Error parsing reconnect timeout: %s", err)
			err = nil

			c.ReconnectTimeout = defaultReconnectTimeout
			log.Printf("[Conf] Using default reconnect_timeout of %v", c.ReconnectTimeout)
		}
	}

	// fetch the remote configuration
	err = c.syncConfig()
	if err != nil {
		log.Printf("[Conf] Error fetching remote config: %s", err)
		return
	}

	c.Log("Services: %s", c.Services)
	c.Log("Groups: %s", c.GroupToServices)

	// initialize the task queues
	c.NewTasks = make(chan task)
	c.DoneTasks = make(chan taskExecution)

	return
}

func main() {
	var err error

	// get arguments
	kingpin.MustParse(app.Parse(os.Args[1:]))

	if *debugAll {
		*brokerDebugging = true
		*scriptDebugging = true
		*backendDebugging = true
	}

	if *brokerDebugging {
		log.Printf("[Debug] broker debugging enabled")
	}
	if *scriptDebugging {
		log.Printf("[Debug] script debugging enabled")
	}
	if *backendDebugging {
		log.Printf("[Debug] backend debugging enabled")
	}
	if *sequentialExecution {
		log.Printf("[Debug] Executing tasks sequentially")
	}

	// read the config file
	c, err := getConfig(*configFile)
	if err != nil {
		log.Fatalf("[Exit] No valid config. Exiting")
	}

	if len(c.Services) == 0 && len(c.GroupToServices) == 0 {
		log.Printf("[P] Not starting pubsub because the are no services to subscribe to")
		return
	}

	// start task handler and responder
	go c.taskHandler()
	go c.taskResponder()

	consumer := c.consumer()
	defer consumer.close()

	consumer.startConsuming()

	// start the fetcher after the consuming starts
	// -> we miss nothing
	go c.taskFetcher()

	// run till killed
	forever := make(chan bool)
	<-forever
}