<commit_before>package connector

import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"strings"
	"time"

	"github.com/facette/facette/pkg/types"
	"github.com/facette/facette/pkg/utils"
)

const (
	graphiteURLMetrics string = "/metrics/index.json"
	graphiteURLRender  string = "/render"
)

type graphitePlot struct {
	Target     string
	Datapoints [][2]float64
}

// GraphiteConnector represents the main structure of the Graphite connector.
type GraphiteConnector struct {
	URL         string
	InsecureTLS bool
	outputChan  *chan [2]string
}

func init() {
	Connectors["graphite"] = func(outputChan *chan [2]string, config map[string]interface{}) (interface{}, error) {
		var (
			configURL           string
			configAllowInsecure string
			ok                  bool
		)

		if _, ok := config["url"]; !ok {
			return nil, fmt.Errorf("missing `url' mandatory connector setting")
		}

		if configURL, ok = config["url"].(string); !ok {
			return nil, fmt.Errorf("connector setting `url' value should be a string")
		}

		if _, ok = config["allow_insecure_tls"]; ok {
			if configAllowInsecure, ok = config["allow_insecure_tls"].(string); !ok {
				return nil, fmt.Errorf("connector setting `allow_insecure_tls' value should be a string")
			}
		}

		connector := &GraphiteConnector{
			URL:         configURL,
			InsecureTLS: configAllowInsecure == "yes",
			outputChan:  outputChan,
		}

		return connector, nil
	}
}

// GetPlots retrieves time series data from origin based on a query and a time interval.
func (connector *GraphiteConnector) GetPlots(query *PlotQuery) (map[string]*PlotResult, error) {
	result := make(map[string]*PlotResult)

	httpTransport := &http.Transport{}
	if connector.InsecureTLS {
		httpTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	}

	httpClient := http.Client{Transport: httpTransport}

	serieName, queryURL, err := graphiteBuildQueryURL(query.Group, query.StartTime, query.EndTime)
	if err != nil {
		return nil, fmt.Errorf("unable to build Graphite query URL: %s", err.Error())
	}

	response, err := httpClient.Get(strings.TrimSuffix(connector.URL, "/") + queryURL)
	if err != nil {
		return nil, err
	}

	if err = graphiteCheckConnectorResponse(response); err != nil {
		return nil, fmt.Errorf("invalid HTTP backend response: %s", err)
	}

	data, err := ioutil.ReadAll(response.Body)
	if err != nil {
		return nil, fmt.Errorf("unable to read HTTP response body: %s", err)
	}

	graphitePlots := make([]graphitePlot, 0)
	if err = json.Unmarshal(data, &graphitePlots); err != nil {
		return nil, fmt.Errorf("unable to unmarshal JSON data: %s", err)
	}

	if result[serieName], err = graphiteExtractPlotResult(graphitePlots); err != nil {
		return nil, fmt.Errorf("unable to extract plot values from backend response: %s", err)
	}

	return result, nil
}

// Refresh triggers a full connector data update.
func (connector *GraphiteConnector) Refresh(errChan chan error) {
	defer close(*connector.outputChan)
	defer close(errChan)

	httpTransport := &http.Transport{
		Dial: (&net.Dialer{
			// Enable dual IPv4/IPv6 stack connectivity:
			DualStack: true,
			// Enforce HTTP connection timeout:
			Timeout: 10 * time.Second, // TODO: parametrize this into configuration setting
		}).Dial,
	}

	if connector.InsecureTLS {
		httpTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	}

	httpClient := http.Client{Transport: httpTransport}

	response, err := httpClient.Get(strings.TrimSuffix(connector.URL, "/") + graphiteURLMetrics)
	if err != nil {
		errChan <- err
		return
	}

	if err = graphiteCheckConnectorResponse(response); err != nil {
		errChan <- fmt.Errorf("invalid HTTP backend response: %s", err)
		return
	}

	data, err := ioutil.ReadAll(response.Body)
	if err != nil {
		errChan <- fmt.Errorf("unable to read HTTP response body: %s", err)
		return
	}

	metrics := make([]string, 0)
	if err = json.Unmarshal(data, &metrics); err != nil {
		errChan <- fmt.Errorf("unable to unmarshal JSON data: %s", err)
		return
	}

	for _, metric := range metrics {
		var sourceName, metricName string

		index := strings.Index(metric, ".")

		if index == -1 {
			// TODO: fix?
			sourceName = "<unknown>"
			metricName = metric
		} else {
			sourceName = metric[0:index]
			metricName = metric[index+1:]
		}

		*connector.outputChan <- [2]string{sourceName, metricName}
	}
}

func graphiteCheckConnectorResponse(response *http.Response) error {
	if response.StatusCode != 200 {
		return fmt.Errorf("got HTTP status code %d, expected 200", response.StatusCode)
	}

	if utils.HTTPGetContentType(response) != "application/json" {
		return fmt.Errorf("got HTTP content type `%s', expected `application/json'", response.Header["Content-Type"])
	}

	return nil
}

func graphiteBuildQueryURL(query *GroupQuery, startTime, endTime time.Time) (string, string, error) {
	var (
		serieName string
		target    string
	)

	now := time.Now()

	fromTime := 0

	queryURL := fmt.Sprintf("%s?format=json", graphiteURLRender)

	if query.Type == OperGroupTypeNone {
		serieName = query.Series[0].Name
		target = fmt.Sprintf("%s.%s", query.Series[0].Metric.SourceName, query.Series[0].Metric.Name)
	} else {
		serieName = query.Name
		targets := make([]string, 0)

		for _, s := range query.Series {
			targets = append(targets, fmt.Sprintf("%s.%s", s.Metric.SourceName, s.Metric.Name))
		}

		target = fmt.Sprintf("group(%s)", strings.Join(targets, ","))

		switch query.Type {
		case OperGroupTypeAvg:
			target = fmt.Sprintf("averageSeries(%s)", target)
		case OperGroupTypeSum:
			target = fmt.Sprintf("sumSeries(%s)", target)
		}
	}

	target = fmt.Sprintf("legendValue(%s, 'min', 'max', 'avg', 'last')", target)

	queryURL += fmt.Sprintf("&target=%s", target)

	if startTime.Before(now) {
		fromTime = int(now.Sub(startTime).Seconds())
	}

	queryURL += fmt.Sprintf("&from=-%ds", fromTime)

	// Only specify `until' parameter if endTime is still in the past
	if endTime.Before(now) {
		untilTime := int(time.Now().Sub(endTime).Seconds())
		queryURL += fmt.Sprintf("&until=-%ds", untilTime)
	}

	return serieName, queryURL, nil
}

func graphiteExtractPlotResult(plots []graphitePlot) (*PlotResult, error) {
	var min, max, avg, last float64

	result := &PlotResult{Info: make(map[string]types.PlotValue)}

	// Return an empty plotResult if Graphite API didn't return any datapoint matching the query
	if len(plots) == 0 || len(plots[0].Datapoints) == 0 {
		return result, nil
	}

	for _, plotPoint := range plots[0].Datapoints {
		result.Plots = append(result.Plots, types.PlotValue(plotPoint[0]))
	}

	// Scan the target legend for plot min/max/avg/last info
	if index := strings.Index(plots[0].Target, "(min"); index > 0 {
		fmt.Sscanf(plots[0].Target[index:], "(min: %f) (max: %f) (avg: %f) (last: %f)", &min, &max, &avg, &last)
	}

	result.Info["min"] = types.PlotValue(min)
	result.Info["max"] = types.PlotValue(max)
	result.Info["avg"] = types.PlotValue(avg)
	result.Info["last"] = types.PlotValue(last)

	return result, nil
}
<commit_msg>Use config settings helpers in Graphite connector<commit_after>package connector

import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"strings"
	"time"

	"github.com/facette/facette/pkg/config"
	"github.com/facette/facette/pkg/types"
	"github.com/facette/facette/pkg/utils"
)

const (
	graphiteURLMetrics string = "/metrics/index.json"
	graphiteURLRender  string = "/render"
)

type graphitePlot struct {
	Target     string
	Datapoints [][2]float64
}

// GraphiteConnector represents the main structure of the Graphite connector.
type GraphiteConnector struct {
	URL         string
	InsecureTLS bool
	outputChan  *chan [2]string
}

func init() {
	Connectors["graphite"] = func(outputChan *chan [2]string, settings map[string]interface{}) (interface{}, error) {
		var err error

		connector := &GraphiteConnector{
			InsecureTLS: false,
			outputChan:  outputChan,
		}

		if connector.URL, err = config.GetString(settings, "url", true); err != nil {
			return nil, err
		}

		if connector.InsecureTLS, err = config.GetBool(settings, "allow_insecure_tls", false); err != nil {
			return nil, err
		}

		return connector, nil
	}
}

// GetPlots retrieves time series data from origin based on a query and a time interval.
func (connector *GraphiteConnector) GetPlots(query *PlotQuery) (map[string]*PlotResult, error) {
	result := make(map[string]*PlotResult)

	httpTransport := &http.Transport{}
	if connector.InsecureTLS {
		httpTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	}

	httpClient := http.Client{Transport: httpTransport}

	serieName, queryURL, err := graphiteBuildQueryURL(query.Group, query.StartTime, query.EndTime)
	if err != nil {
		return nil, fmt.Errorf("unable to build Graphite query URL: %s", err.Error())
	}

	response, err := httpClient.Get(strings.TrimSuffix(connector.URL, "/") + queryURL)
	if err != nil {
		return nil, err
	}

	if err = graphiteCheckConnectorResponse(response); err != nil {
		return nil, fmt.Errorf("invalid HTTP backend response: %s", err)
	}

	data, err := ioutil.ReadAll(response.Body)
	if err != nil {
		return nil, fmt.Errorf("unable to read HTTP response body: %s", err)
	}

	graphitePlots := make([]graphitePlot, 0)
	if err = json.Unmarshal(data, &graphitePlots); err != nil {
		return nil, fmt.Errorf("unable to unmarshal JSON data: %s", err)
	}

	if result[serieName], err = graphiteExtractPlotResult(graphitePlots); err != nil {
		return nil, fmt.Errorf("unable to extract plot values from backend response: %s", err)
	}

	return result, nil
}

// Refresh triggers a full connector data update.
func (connector *GraphiteConnector) Refresh(errChan chan error) {
	defer close(*connector.outputChan)
	defer close(errChan)

	httpTransport := &http.Transport{
		Dial: (&net.Dialer{
			// Enable dual IPv4/IPv6 stack connectivity:
			DualStack: true,
			// Enforce HTTP connection timeout:
			Timeout: 10 * time.Second, // TODO: parametrize this into configuration setting
		}).Dial,
	}

	if connector.InsecureTLS {
		httpTransport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	}

	httpClient := http.Client{Transport: httpTransport}

	response, err := httpClient.Get(strings.TrimSuffix(connector.URL, "/") + graphiteURLMetrics)
	if err != nil {
		errChan <- err
		return
	}

	if err = graphiteCheckConnectorResponse(response); err != nil {
		errChan <- fmt.Errorf("invalid HTTP backend response: %s", err)
		return
	}

	data, err := ioutil.ReadAll(response.Body)
	if err != nil {
		errChan <- fmt.Errorf("unable to read HTTP response body: %s", err)
		return
	}

	metrics := make([]string, 0)
	if err = json.Unmarshal(data, &metrics); err != nil {
		errChan <- fmt.Errorf("unable to unmarshal JSON data: %s", err)
		return
	}

	for _, metric := range metrics {
		var sourceName, metricName string

		index := strings.Index(metric, ".")

		if index == -1 {
			// TODO: fix?
			sourceName = "<unknown>"
			metricName = metric
		} else {
			sourceName = metric[0:index]
			metricName = metric[index+1:]
		}

		*connector.outputChan <- [2]string{sourceName, metricName}
	}
}

func graphiteCheckConnectorResponse(response *http.Response) error {
	if response.StatusCode != 200 {
		return fmt.Errorf("got HTTP status code %d, expected 200", response.StatusCode)
	}

	if utils.HTTPGetContentType(response) != "application/json" {
		return fmt.Errorf("got HTTP content type `%s', expected `application/json'", response.Header["Content-Type"])
	}

	return nil
}

func graphiteBuildQueryURL(query *GroupQuery, startTime, endTime time.Time) (string, string, error) {
	var (
		serieName string
		target    string
	)

	now := time.Now()

	fromTime := 0

	queryURL := fmt.Sprintf("%s?format=json", graphiteURLRender)

	if query.Type == OperGroupTypeNone {
		serieName = query.Series[0].Name
		target = fmt.Sprintf("%s.%s", query.Series[0].Metric.SourceName, query.Series[0].Metric.Name)
	} else {
		serieName = query.Name
		targets := make([]string, 0)

		for _, s := range query.Series {
			targets = append(targets, fmt.Sprintf("%s.%s", s.Metric.SourceName, s.Metric.Name))
		}

		target = fmt.Sprintf("group(%s)", strings.Join(targets, ","))

		switch query.Type {
		case OperGroupTypeAvg:
			target = fmt.Sprintf("averageSeries(%s)", target)
		case OperGroupTypeSum:
			target = fmt.Sprintf("sumSeries(%s)", target)
		}
	}

	target = fmt.Sprintf("legendValue(%s, 'min', 'max', 'avg', 'last')", target)

	queryURL += fmt.Sprintf("&target=%s", target)

	if startTime.Before(now) {
		fromTime = int(now.Sub(startTime).Seconds())
	}

	queryURL += fmt.Sprintf("&from=-%ds", fromTime)

	// Only specify `until' parameter if endTime is still in the past
	if endTime.Before(now) {
		untilTime := int(time.Now().Sub(endTime).Seconds())
		queryURL += fmt.Sprintf("&until=-%ds", untilTime)
	}

	return serieName, queryURL, nil
}

func graphiteExtractPlotResult(plots []graphitePlot) (*PlotResult, error) {
	var min, max, avg, last float64

	result := &PlotResult{Info: make(map[string]types.PlotValue)}

	// Return an empty plotResult if Graphite API didn't return any datapoint matching the query
	if len(plots) == 0 || len(plots[0].Datapoints) == 0 {
		return result, nil
	}

	for _, plotPoint := range plots[0].Datapoints {
		result.Plots = append(result.Plots, types.PlotValue(plotPoint[0]))
	}

	// Scan the target legend for plot min/max/avg/last info
	if index := strings.Index(plots[0].Target, "(min"); index > 0 {
		fmt.Sscanf(plots[0].Target[index:], "(min: %f) (max: %f) (avg: %f) (last: %f)", &min, &max, &avg, &last)
	}

	result.Info["min"] = types.PlotValue(min)
	result.Info["max"] = types.PlotValue(max)
	result.Info["avg"] = types.PlotValue(avg)
	result.Info["last"] = types.PlotValue(last)

	return result, nil
}

{'fraction_non_alphanumeric': 0.14254497778425232, 'fraction_numerical': 0.0052443732245611475, 'mean_word_length': 4.280769230769231, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 7, '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': '5292689', 'n_tokens_mistral': 4941, 'n_tokens_neox': 4515, 'n_words': 1167}