<commit_before>/*******************************************************************************
*
* Copyright 2017-2020 SAP SE
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You should have received a copy of the License along with this
* program. If not, 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"
	"errors"
	"fmt"
	"net/http"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	policy "github.com/databus23/goslo.policy"
	"github.com/dlmiddlecote/sqlstats"
	"github.com/gophercloud/gophercloud"
	"github.com/gorilla/mux"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/rs/cors"
	"github.com/sapcc/go-bits/gopherpolicy"
	"github.com/sapcc/go-bits/httpee"
	"github.com/sapcc/go-bits/logg"
	"github.com/sapcc/go-bits/respondwith"
	"gopkg.in/yaml.v2"

	"github.com/sapcc/limes/pkg/api"
	"github.com/sapcc/limes/pkg/collector"
	"github.com/sapcc/limes/pkg/core"
	"github.com/sapcc/limes/pkg/db"
	"github.com/sapcc/limes/pkg/util"

	_ "github.com/sapcc/limes/pkg/plugins"
)

var discoverInterval = 3 * time.Minute

func main() {
	//first two arguments must be task name and configuration file
	if len(os.Args) < 3 {
		printUsageAndExit()
	}
	taskName, configPath, remainingArgs := os.Args[1], os.Args[2], os.Args[3:]
	util.Component = "limes-" + taskName

	//load configuration and connect to cluster
	cluster := core.NewConfiguration(configPath)
	err := cluster.Connect()
	if err != nil {
		logg.Fatal(util.ErrorToString(err))
	}
	api.StartAuditTrail(cluster.ID, cluster.Config.CADF)

	//select task
	var task func(*core.Cluster, []string) error
	switch taskName {
	case "collect":
		task = taskCollect
	case "serve":
		task = taskServe
	case "test-get-quota":
		task = taskTestGetQuota
	case "test-get-rates":
		task = taskTestGetRates
	case "test-set-quota":
		task = taskTestSetQuota
	case "test-scan-capacity":
		task = taskTestScanCapacity
	default:
		printUsageAndExit()
	}

	//run task
	err = task(cluster, remainingArgs)
	if err != nil {
		logg.Fatal(util.ErrorToString(err))
	}
}

var usageMessage = strings.Replace(strings.TrimSpace(`
Usage:
\t%s (collect|serve) <config-file>
\t%s test-get-quota <config-file> <project-id> <service-type>
\t%s test-get-rates <config-file> <project-id> <service-type> [<prev-serialized-state>]
\t%s test-set-quota <config-file> <project-id> <service-type> <resource-name>=<integer-value>...
\t%s test-scan-capacity <config-file> <capacitor>
`), `\t`, "\t", -1) + "\n"

func printUsageAndExit() {
	fmt.Fprintln(os.Stderr, strings.Replace(usageMessage, "%s", os.Args[0], -1))
	os.Exit(1)
}

////////////////////////////////////////////////////////////////////////////////
// task: collect

func taskCollect(cluster *core.Cluster, args []string) error {
	if len(args) != 0 {
		printUsageAndExit()
	}

	//connect to database
	err := db.Init()
	if err != nil {
		logg.Fatal(err.Error())
	}
	prometheus.MustRegister(sqlstats.NewStatsCollector("limes", db.DB.Db))

	//start scraping threads (NOTE: Many people use a pair of sync.WaitGroup and
	//stop channel to shutdown threads in a controlled manner. I decided against
	//that for now, and instead construct worker threads in such a way that they
	//can be terminated at any time without leaving the system in an inconsistent
	//state, mostly through usage of DB transactions.)
	for _, plugin := range cluster.QuotaPlugins {
		c := collector.NewCollector(cluster, plugin)
		go c.Scrape()
		go c.ScrapeRates()
	}

	//start those collector threads which operate over all services simultaneously
	c := collector.NewCollector(cluster, nil)
	go c.CheckConsistency()
	go c.ScanCapacity()
	go func() {
		for {
			_, err := collector.ScanDomains(cluster, collector.ScanDomainsOpts{ScanAllProjects: true})
			if err != nil {
				logg.Error(util.ErrorToString(err))
			}
			time.Sleep(discoverInterval)
		}
	}()

	//use main thread to emit Prometheus metrics
	prometheus.MustRegister(&collector.AggregateMetricsCollector{Cluster: cluster})
	prometheus.MustRegister(&collector.CapacityPluginMetricsCollector{Cluster: cluster})
	prometheus.MustRegister(&collector.QuotaPluginMetricsCollector{Cluster: cluster})
	//nolint:errcheck
	if exposeMetrics, _ := strconv.ParseBool(os.Getenv("LIMES_COLLECTOR_DATA_METRICS_EXPOSE")); exposeMetrics {
		skipZero, _ := strconv.ParseBool(os.Getenv("LIMES_COLLECTOR_DATA_METRICS_SKIP_ZERO"))
		prometheus.MustRegister(&collector.DataMetricsCollector{
			Cluster:      cluster,
			ReportZeroes: !skipZero,
		})
	}
	http.Handle("/metrics", promhttp.Handler())
	metricsListenAddr := util.EnvOrDefault("LIMES_COLLECTOR_METRICS_LISTEN_ADDRESS", ":8080")
	logg.Info("listening on " + metricsListenAddr)
	return httpee.ListenAndServeContext(httpee.ContextWithSIGINT(context.Background(), 10*time.Second), metricsListenAddr, nil)
}

////////////////////////////////////////////////////////////////////////////////
// task: serve

func taskServe(cluster *core.Cluster, args []string) error {
	if len(args) != 0 {
		printUsageAndExit()
	}

	//connect to database
	err := db.Init()
	if err != nil {
		logg.Fatal(err.Error())
	}
	prometheus.MustRegister(sqlstats.NewStatsCollector("limes", db.DB.Db))

	//load oslo.policy file
	policyEnforcer, err := loadPolicyFile(util.EnvOrDefault("LIMES_API_POLICY_PATH", "/etc/limes/policy.yaml"))
	if err != nil {
		logg.Fatal("could not load policy file: %s", err.Error())
	}

	mainRouter := mux.NewRouter()

	//hook up the v1 API (this code is structured so that a newer API version can
	//be added easily later)
	v1Router, v1VersionData := api.NewV1Router(cluster, policyEnforcer)
	mainRouter.PathPrefix("/v1/").Handler(v1Router)

	//add the version advertisement that lists all available API versions
	mainRouter.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		allVersions := struct {
			Versions []api.VersionData `json:"versions"`
		}{[]api.VersionData{v1VersionData}}
		respondwith.JSON(w, 300, allVersions)
	})
	var handler http.Handler = mainRouter

	//add Prometheus instrumentation
	http.Handle("/metrics", promhttp.Handler())

	//add logging instrumentation
	exceptCodeStrings := strings.Split(os.Getenv("LIMES_API_REQUEST_LOG_EXCEPT_STATUS_CODES"), ",")
	var exceptCodes []int
	for _, v := range exceptCodeStrings {
		code, err := strconv.Atoi(strings.TrimSpace(v))
		if err != nil {
			logg.Fatal("could not parse LIMES_API_REQUEST_LOG_EXCEPT_STATUS_CODES: %s", err.Error())
		}
		exceptCodes = append(exceptCodes, code)
	}
	handler = logg.Middleware{ExceptStatusCodes: exceptCodes}.Wrap(handler)

	//add CORS support
	allowedOriginStr := strings.ReplaceAll(os.Getenv("LIMES_API_CORS_ALLOWED_ORIGINS"), " ", "")
	allowedOrigins := strings.Split(allowedOriginStr, "||")
	if len(allowedOrigins) > 0 {<commit_msg>fix access to /rates/v1 endpoints<commit_after>/*******************************************************************************
*
* Copyright 2017-2020 SAP SE
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You should have received a copy of the License along with this
* program. If not, 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"
	"errors"
	"fmt"
	"net/http"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	policy "github.com/databus23/goslo.policy"
	"github.com/dlmiddlecote/sqlstats"
	"github.com/gophercloud/gophercloud"
	"github.com/gorilla/mux"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/rs/cors"
	"github.com/sapcc/go-bits/gopherpolicy"
	"github.com/sapcc/go-bits/httpee"
	"github.com/sapcc/go-bits/logg"
	"github.com/sapcc/go-bits/respondwith"
	"gopkg.in/yaml.v2"

	"github.com/sapcc/limes/pkg/api"
	"github.com/sapcc/limes/pkg/collector"
	"github.com/sapcc/limes/pkg/core"
	"github.com/sapcc/limes/pkg/db"
	"github.com/sapcc/limes/pkg/util"

	_ "github.com/sapcc/limes/pkg/plugins"
)

var discoverInterval = 3 * time.Minute

func main() {
	//first two arguments must be task name and configuration file
	if len(os.Args) < 3 {
		printUsageAndExit()
	}
	taskName, configPath, remainingArgs := os.Args[1], os.Args[2], os.Args[3:]
	util.Component = "limes-" + taskName

	//load configuration and connect to cluster
	cluster := core.NewConfiguration(configPath)
	err := cluster.Connect()
	if err != nil {
		logg.Fatal(util.ErrorToString(err))
	}
	api.StartAuditTrail(cluster.ID, cluster.Config.CADF)

	//select task
	var task func(*core.Cluster, []string) error
	switch taskName {
	case "collect":
		task = taskCollect
	case "serve":
		task = taskServe
	case "test-get-quota":
		task = taskTestGetQuota
	case "test-get-rates":
		task = taskTestGetRates
	case "test-set-quota":
		task = taskTestSetQuota
	case "test-scan-capacity":
		task = taskTestScanCapacity
	default:
		printUsageAndExit()
	}

	//run task
	err = task(cluster, remainingArgs)
	if err != nil {
		logg.Fatal(util.ErrorToString(err))
	}
}

var usageMessage = strings.Replace(strings.TrimSpace(`
Usage:
\t%s (collect|serve) <config-file>
\t%s test-get-quota <config-file> <project-id> <service-type>
\t%s test-get-rates <config-file> <project-id> <service-type> [<prev-serialized-state>]
\t%s test-set-quota <config-file> <project-id> <service-type> <resource-name>=<integer-value>...
\t%s test-scan-capacity <config-file> <capacitor>
`), `\t`, "\t", -1) + "\n"

func printUsageAndExit() {
	fmt.Fprintln(os.Stderr, strings.Replace(usageMessage, "%s", os.Args[0], -1))
	os.Exit(1)
}

////////////////////////////////////////////////////////////////////////////////
// task: collect

func taskCollect(cluster *core.Cluster, args []string) error {
	if len(args) != 0 {
		printUsageAndExit()
	}

	//connect to database
	err := db.Init()
	if err != nil {
		logg.Fatal(err.Error())
	}
	prometheus.MustRegister(sqlstats.NewStatsCollector("limes", db.DB.Db))

	//start scraping threads (NOTE: Many people use a pair of sync.WaitGroup and
	//stop channel to shutdown threads in a controlled manner. I decided against
	//that for now, and instead construct worker threads in such a way that they
	//can be terminated at any time without leaving the system in an inconsistent
	//state, mostly through usage of DB transactions.)
	for _, plugin := range cluster.QuotaPlugins {
		c := collector.NewCollector(cluster, plugin)
		go c.Scrape()
		go c.ScrapeRates()
	}

	//start those collector threads which operate over all services simultaneously
	c := collector.NewCollector(cluster, nil)
	go c.CheckConsistency()
	go c.ScanCapacity()
	go func() {
		for {
			_, err := collector.ScanDomains(cluster, collector.ScanDomainsOpts{ScanAllProjects: true})
			if err != nil {
				logg.Error(util.ErrorToString(err))
			}
			time.Sleep(discoverInterval)
		}
	}()

	//use main thread to emit Prometheus metrics
	prometheus.MustRegister(&collector.AggregateMetricsCollector{Cluster: cluster})
	prometheus.MustRegister(&collector.CapacityPluginMetricsCollector{Cluster: cluster})
	prometheus.MustRegister(&collector.QuotaPluginMetricsCollector{Cluster: cluster})
	//nolint:errcheck
	if exposeMetrics, _ := strconv.ParseBool(os.Getenv("LIMES_COLLECTOR_DATA_METRICS_EXPOSE")); exposeMetrics {
		skipZero, _ := strconv.ParseBool(os.Getenv("LIMES_COLLECTOR_DATA_METRICS_SKIP_ZERO"))
		prometheus.MustRegister(&collector.DataMetricsCollector{
			Cluster:      cluster,
			ReportZeroes: !skipZero,
		})
	}
	http.Handle("/metrics", promhttp.Handler())
	metricsListenAddr := util.EnvOrDefault("LIMES_COLLECTOR_METRICS_LISTEN_ADDRESS", ":8080")
	logg.Info("listening on " + metricsListenAddr)
	return httpee.ListenAndServeContext(httpee.ContextWithSIGINT(context.Background(), 10*time.Second), metricsListenAddr, nil)
}

////////////////////////////////////////////////////////////////////////////////
// task: serve

func taskServe(cluster *core.Cluster, args []string) error {
	if len(args) != 0 {
		printUsageAndExit()
	}

	//connect to database
	err := db.Init()
	if err != nil {
		logg.Fatal(err.Error())
	}
	prometheus.MustRegister(sqlstats.NewStatsCollector("limes", db.DB.Db))

	//load oslo.policy file
	policyEnforcer, err := loadPolicyFile(util.EnvOrDefault("LIMES_API_POLICY_PATH", "/etc/limes/policy.yaml"))
	if err != nil {
		logg.Fatal("could not load policy file: %s", err.Error())
	}

	mainRouter := mux.NewRouter()

	//hook up the v1 API (this code is structured so that a newer API version can
	//be added easily later)
	v1Router, v1VersionData := api.NewV1Router(cluster, policyEnforcer)
	mainRouter.PathPrefix("/v1/").Handler(v1Router)
	mainRouter.PathPrefix("/rates/v1/").Handler(v1Router)

	//add the version advertisement that lists all available API versions
	mainRouter.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		allVersions := struct {
			Versions []api.VersionData `json:"versions"`
		}{[]api.VersionData{v1VersionData}}
		respondwith.JSON(w, 300, allVersions)
	})
	var handler http.Handler = mainRouter

	//add Prometheus instrumentation
	http.Handle("/metrics", promhttp.Handler())

	//add logging instrumentation
	exceptCodeStrings := strings.Split(os.Getenv("LIMES_API_REQUEST_LOG_EXCEPT_STATUS_CODES"), ",")
	var exceptCodes []int
	for _, v := range exceptCodeStrings {
		code, err := strconv.Atoi(strings.TrimSpace(v))
		if err != nil {
			logg.Fatal("could not parse LIMES_API_REQUEST_LOG_EXCEPT_STATUS_CODES: %s", err.Error())
		}
		exceptCodes = append(exceptCodes, code)
	}
	handler = logg.Middleware{ExceptStatusCodes: exceptCodes}.Wrap(handler)

	//add CORS support
	allowedOriginStr := strings.ReplaceAll(os.Getenv("LIMES_API_CORS_ALLOWED_ORIGINS"), " ", "")
	allowedOrigins := strings.Split(allowedOriginStr, "||")
	if len(allowedOrigins) > 0 {
{'fraction_non_alphanumeric': 0.17011322801545867, 'fraction_numerical': 0.00603430741067191, 'mean_word_length': 4.282951289398281, 'pattern_counts': {'":': 14, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '7066437', 'n_tokens_mistral': 5104, 'n_tokens_neox': 4824, 'n_words': 1121}