// Copyright The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// 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 (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/prometheus/prometheus/util/testutil"
)

type origin int

const (
	apiOrigin origin = iota
	consoleOrigin
	ruleOrigin
)

// queryLogTest defines a query log test.
type queryLogTest struct {
	origin         origin   // Kind of queries tested: api, console, rules.
	prefix         string   // Set as --web.route-prefix.
	host           string   // Used in --web.listen-address. Used with 127.0.0.1 and ::1.
	port           int      // Used in --web.listen-address.
	cwd            string   // Directory where the test is running. Required to find the rules in testdata.
	configFile     *os.File // The configuration file.
	enabledAtStart bool     // Whether query log is enabled at startup.
}

// skip checks if the test is needed and the prerequisites are met.
func (p *queryLogTest) skip(t *testing.T) {
	if p.prefix != "" && p.origin == ruleOrigin {
		t.Skip("changing prefix has no effect on rules")
	}
	// Some systems don't support IPv4 or IPv6.
	l, err := net.Listen("tcp", fmt.Sprintf("%s:0", p.host))
	if err != nil {
		t.Skip("ip version not supported")
	}
	l.Close()
}

// waitForPrometheus waits for Prometheus to be ready.
func (p *queryLogTest) waitForPrometheus(t *testing.T) {
	t.Helper()
	require.Eventually(t, func() bool {
		r, err := http.Get(fmt.Sprintf("http://%s:%d%s/-/ready", p.host, p.port, p.prefix))
		if err != nil {
			return false
		}
		r.Body.Close()
		return r.StatusCode == http.StatusOK
	}, 20*time.Second, 500*time.Millisecond, "prometheus at %s:%d did not become ready in time", p.host, p.port)
}

// setQueryLog alters the configuration file to enable or disable the query log,
// then reloads the configuration if needed.
func (p *queryLogTest) setQueryLog(t *testing.T, queryLogFile string) {
	err := p.configFile.Truncate(0)
	require.NoError(t, err)
	_, err = p.configFile.Seek(0, 0)
	require.NoError(t, err)
	if queryLogFile != "" {
		_, err = fmt.Fprintf(p.configFile, "global:\n  query_log_file: %s\n", queryLogFile)
		require.NoError(t, err)
	}
	_, err = p.configFile.Write([]byte(p.configuration()))
	require.NoError(t, err)
}

// query runs a query according to the test origin.
func (p *queryLogTest) query(t *testing.T) {
	switch p.origin {
	case apiOrigin:
		r, err := http.Get(fmt.Sprintf(
			"http://%s:%d%s/api/v1/query_range?step=5&start=0&end=3600&query=%s",
			p.host,
			p.port,
			p.prefix,
			url.QueryEscape("query_with_api"),
		))
		require.NoError(t, err)
		require.Equal(t, 200, r.StatusCode)
	case consoleOrigin:
		r, err := http.Get(fmt.Sprintf(
			"http://%s:%d%s/consoles/test.html",
			p.host,
			p.port,
			p.prefix,
		))
		require.NoError(t, err)
		require.Equal(t, 200, r.StatusCode)
	case ruleOrigin:
		// Poll the /api/v1/rules endpoint until a new rule evaluation is detected.
		var lastEvalTime time.Time
		for {
			r, err := http.Get(fmt.Sprintf("http://%s:%d/api/v1/rules", p.host, p.port))
			require.NoError(t, err)

			rulesBody, err := io.ReadAll(r.Body)
			require.NoError(t, err)
			defer r.Body.Close()

			// Parse the rules response to find the last evaluation time.
			newEvalTime := parseLastEvaluation(rulesBody)
			if newEvalTime.After(lastEvalTime) {
				if !lastEvalTime.IsZero() {
					break
				}
				lastEvalTime = newEvalTime
			}

			time.Sleep(100 * time.Millisecond)
		}
	default:
		panic("can't query this origin")
	}
}

// parseLastEvaluation extracts the last evaluation timestamp from the /api/v1/rules response.
func parseLastEvaluation(rulesBody []byte) time.Time {
	var ruleResponse struct {
		Status string `json:"status"`
		Data   struct {
			Groups []struct {
				Rules []struct {
					LastEvaluation string `json:"lastEvaluation"`
				} `json:"rules"`
			} `json:"groups"`
		} `json:"data"`
	}

	err := json.Unmarshal(rulesBody, &ruleResponse)
	if err != nil {
		return time.Time{}
	}

	for _, group := range ruleResponse.Data.Groups {
		for _, rule := range group.Rules {
			if evalTime, err := time.Parse(time.RFC3339Nano, rule.LastEvaluation); err == nil {
				return evalTime
			}
		}
	}

	return time.Time{}
}

// queryString returns the expected queryString of a this test.
func (p *queryLogTest) queryString() string {
	switch p.origin {
	case apiOrigin:
		return "query_with_api"
	case ruleOrigin:
		return "query_in_rule"
	case consoleOrigin:
		return "query_in_console"
	default:
		panic("unknown origin")
	}
}

// validateLastQuery checks that the last query in the query log matches the
// test parameters.
func (p *queryLogTest) validateLastQuery(t *testing.T, ql []queryLogLine) {
	q := ql[len(ql)-1]
	require.Equal(t, p.queryString(), q.Params.Query)

	switch p.origin {
	case apiOrigin:
		require.Equal(t, 5, q.Params.Step)
		require.Equal(t, "1970-01-01T00:00:00.000Z", q.Params.Start)
		require.Equal(t, "1970-01-01T01:00:00.000Z", q.Params.End)
	default:
		require.Equal(t, 0, q.Params.Step)
	}

	if p.origin != ruleOrigin {
		host := p.host
		if host == "[::1]" {
			host = "::1"
		}
		require.Equal(t, host, q.Request.ClientIP)
	}

	switch p.origin {
	case apiOrigin:
		require.Equal(t, p.prefix+"/api/v1/query_range", q.Request.Path)
	case consoleOrigin:
		require.Equal(t, p.prefix+"/consoles/test.html", q.Request.Path)
	case ruleOrigin:
		require.Equal(t, "querylogtest", q.RuleGroup.Name)
		require.Equal(t, filepath.Join(p.cwd, "testdata", "rules", "test.yml"), q.RuleGroup.File)
	default:
		panic("unknown origin")
	}
}

func (p *queryLogTest) String() string {
	var name string
	switch p.origin {
	case apiOrigin:
		name = "api queries"
	case consoleOrigin:
		name = "console queries"
	case ruleOrigin:
		name = "rule queries"
	}
	name = name + ", " + p.host + ":" + strconv.Itoa(p.port)
	if p.enabledAtStart {
		name += ", enabled at start"
	}
	if p.prefix != "" {
		name = name + ", with prefix " + p.prefix
	}
	return name
}

// params returns the specific command line parameters of this test.
func (p *queryLogTest) params() []string {
	s := []string{}
	if p.prefix != "" {
		s = append(s, "--web.route-prefix="+p.prefix)
	}
	if p.origin == consoleOrigin {
		s = append(s, "--web.console.templates="+filepath.Join(p.cwd, "testdata", "consoles"))
	}
	return s
}

// configuration returns the specific configuration lines required for this
// test.
func (p *queryLogTest) configuration() string {
	switch p.origin {
	case ruleOrigin:
		return "\nrule_files:\n- " + filepath.Join(p.cwd, "testdata", "rules", "test.yml") + "\n"
	default:
		return "\n"
	}
}

// exactQueryCount returns whether we can match an exact query count. False on
// recording rules are they are regular time intervals.
func (p *queryLogTest) exactQueryCount() bool {
	return p.origin != ruleOrigin
}

// run launches the scenario of this query log test.
func (p *queryLogTest) run(t *testing.T) {
	p.skip(t)

	// Setup temporary files for this test.
	queryLogFile, err := os.CreateTemp("", "query")
	require.NoError(t, err)
	defer os.Remove(queryLogFile.Name())
	p.configFile, err = os.CreateTemp("", "config")
	require.NoError(t, err)
	defer os.Remove(p.configFile.Name())

	if p.enabledAtStart {
		p.setQueryLog(t, queryLogFile.Name())
	} else {
		p.setQueryLog(t, "")
	}

	dir := t.TempDir()

	params := append([]string{
		"-test.main",
		"--config.file=" + p.configFile.Name(),
		"--web.enable-lifecycle",
		fmt.Sprintf("--web.listen-address=%s:%d", p.host, p.port),
		"--storage.tsdb.path=" + dir,
	}, p.params()...)

	prom := exec.Command(promPath, params...)
	reloadURL := fmt.Sprintf("http://%s:%d%s/-/reload", p.host, p.port, p.prefix)

	// Log stderr in case of failure.
	stderr, err := prom.StderrPipe()
	require.NoError(t, err)

	// We use a WaitGroup to avoid calling t.Log after the test is done.
	var wg sync.WaitGroup
	wg.Add(1)
	defer wg.Wait()
	go func() {
		slurp, _ := io.ReadAll(stderr)
		t.Log(string(slurp))
		wg.Done()
	}()

	require.NoError(t, prom.Start())

	defer func() {
		prom.Process.Kill()
		prom.Wait()
	}()
	p.waitForPrometheus(t)

	if !p.enabledAtStart {
		p.query(t)
		require.Empty(t, readQueryLog(t, queryLogFile.Name()))
		p.setQueryLog(t, queryLogFile.Name())
		reloadPrometheusConfig(t, reloadURL)
	}

	p.query(t)

	// Wait for query log entry to be written (avoid race with file I/O).
	ql := waitForQueryLog(t, queryLogFile.Name(), 1)
	qc := len(ql)
	if p.exactQueryCount() {
		require.Equal(t, 1, qc)
	} else {
		require.Positive(t, qc, "no queries logged")
	}
	p.validateLastQuery(t, ql)

	p.setQueryLog(t, "")
	reloadPrometheusConfig(t, reloadURL)
	if !p.exactQueryCount() {
		qc = len(readQueryLog(t, queryLogFile.Name()))
	}

	p.query(t)

	ql = readQueryLog(t, queryLogFile.Name())
	require.Len(t, ql, qc)

	qc = len(ql)
	p.setQueryLog(t, queryLogFile.Name())
	reloadPrometheusConfig(t, reloadURL)

	p.query(t)
	qc++

	// Wait for query log entry to be written (avoid race with file I/O).
	ql = waitForQueryLog(t, queryLogFile.Name(), qc)
	if p.exactQueryCount() {
		require.Len(t, ql, qc)
	} else {
		require.GreaterOrEqual(t, len(ql), qc, "no queries logged")
	}
	p.validateLastQuery(t, ql)
	qc = len(ql)

	// The last part of the test can not succeed on Windows because you can't
	// rename files used by other processes.
	if runtime.GOOS == "windows" {
		return
	}
	// Move the file, Prometheus should still write to the old file.
	newFile, err := os.CreateTemp("", "newLoc")
	require.NoError(t, err)
	require.NoError(t, newFile.Close())
	defer os.Remove(newFile.Name())
	require.NoError(t, os.Rename(queryLogFile.Name(), newFile.Name()))
	ql = readQueryLog(t, newFile.Name())
	if p.exactQueryCount() {
		require.Len(t, ql, qc)
	}
	p.validateLastQuery(t, ql)
	qc = len(ql)

	p.query(t)

	qc++

	// Wait for query log entry to be written (avoid race with file I/O).
	ql = waitForQueryLog(t, newFile.Name(), qc)
	if p.exactQueryCount() {
		require.Len(t, ql, qc)
	} else {
		require.GreaterOrEqual(t, len(ql), qc, "no queries logged")
	}
	p.validateLastQuery(t, ql)

	reloadPrometheusConfig(t, reloadURL)

	p.query(t)

	// Wait for query log entry to be written (avoid race with file I/O).
	ql = waitForQueryLog(t, queryLogFile.Name(), 1)
	qc = len(ql)
	if p.exactQueryCount() {
		require.Equal(t, 1, qc)
	} else {
		require.Positive(t, qc, "no queries logged")
	}
}

type queryLogLine struct {
	Params struct {
		Query string `json:"query"`
		Step  int    `json:"step"`
		Start string `json:"start"`
		End   string `json:"end"`
	} `json:"params"`
	Request struct {
		Path     string `json:"path"`
		ClientIP string `json:"clientIP"`
	} `json:"httpRequest"`
	RuleGroup struct {
		File string `json:"file"`
		Name string `json:"name"`
	} `json:"ruleGroup"`
}

// readQueryLog unmarshal a json-formatted query log into query log lines.
func readQueryLog(t *testing.T, path string) []queryLogLine {
	ql := []queryLogLine{}
	file, err := os.Open(path)
	require.NoError(t, err)
	defer file.Close()

	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		var q queryLogLine
		require.NoError(t, json.Unmarshal(scanner.Bytes(), &q))
		ql = append(ql, q)
	}
	return ql
}

// waitForQueryLog waits for the query log to contain at least minEntries entries,
// polling at regular intervals until the timeout is reached.
func waitForQueryLog(t *testing.T, path string, minEntries int) []queryLogLine {
	t.Helper()
	var ql []queryLogLine
	require.Eventually(t, func() bool {
		ql = readQueryLog(t, path)
		return len(ql) >= minEntries
	}, 5*time.Second, 100*time.Millisecond, "timed out waiting for query log to have at least %d entries, got %d", minEntries, len(ql))
	return ql
}

func TestQueryLog(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping test in short mode.")
	}
	t.Parallel()

	cwd, err := os.Getwd()
	require.NoError(t, err)

	for _, host := range []string{"127.0.0.1", "[::1]"} {
		for _, prefix := range []string{"", "/foobar"} {
			for _, enabledAtStart := range []bool{true, false} {
				for _, origin := range []origin{apiOrigin, consoleOrigin, ruleOrigin} {
					p := &queryLogTest{
						origin:         origin,
						host:           host,
						enabledAtStart: enabledAtStart,
						prefix:         prefix,
						port:           testutil.RandomUnprivilegedPort(t),
						cwd:            cwd,
					}

					t.Run(p.String(), func(t *testing.T) {
						t.Parallel()
						p.run(t)
					})
				}
			}
		}
	}
}
