package podautoscaler

import (
	"context"
	"fmt"
	"time"

	"github.com/pingcap/errors"
	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	metricsapi "k8s.io/metrics/pkg/apis/metrics/v1beta1"
	resourceclient "k8s.io/metrics/pkg/client/clientset/versioned/typed/metrics/v1beta1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// PodMetric contains pod metric value (the metric values are expected to be the metric as a milli-value)
type PodMetric struct {
	// Timestamp is the time at which the metric was collected.
	Timestamp time.Time
	// Window is the duration over which the metric was collected.
	Window time.Duration
	// Value is the value of the metric in milli-units.
	Value int64
}

// PodMetricsInfo contains pod metrics as a map from pod namespace/name to PodMetricsInfo
type PodMetricsInfo map[client.ObjectKey]PodMetric

func newMetricsClient(resourceClient resourceclient.PodMetricsesGetter) *metricsClient {
	return &metricsClient{
		&resourceMetricsClient{resourceClient},
	}
}

// metricsClient is a client which supports fetching metrics from the resource metrics API, the
// custom metrics API and the external metrics API.
type metricsClient struct {
	*resourceMetricsClient
}

// resourceMetricsClient implements the resource-metrics-related parts of metricsClient,
// using data from the resource metrics API.
type resourceMetricsClient struct {
	client resourceclient.PodMetricsesGetter
}

var ErrNoMetricsFound = errors.New("no metrics returned from resource metrics API")

// GetResourceMetric gets the given resource metric (and an associated oldest timestamp)
// for all pods matching the specified selector in the given namespace
func (c *resourceMetricsClient) GetResourceMetric(ctx context.Context, resource v1.ResourceName, namespace string, selector labels.Selector) (PodMetricsInfo, time.Time, error) {
	metrics, err := c.client.PodMetricses(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector.String()})
	if err != nil {
		return nil, time.Time{}, fmt.Errorf("unable to fetch metrics from resource metrics API: %v", err)
	}

	if len(metrics.Items) == 0 {
		return nil, time.Time{}, ErrNoMetricsFound
	}

	res := getPodMetrics(metrics.Items, resource)
	timestamp := metrics.Items[0].Timestamp.Time
	return res, timestamp, nil
}

// getPodMetrics extracts the given resource metric from the raw pod metrics
// obtained from the resource metrics API.
//
// CPU Metrics:
//
// CPU is reported as the average core usage measured in cpu units. One cpu, in Kubernetes,
// is equivalent to 1 vCPU/Core for cloud providers, and 1 hyper-thread on bare-metal Intel processors.
// This value is derived by taking a rate over a cumulative CPU counter provided
// by the kernel (in both Linux and Windows kernels).
// The time window used to calculate CPU is shown under window field in Metrics API.
//
// Memory Metrics:
//
// Memory is reported as the working set, measured in bytes, at the instant the metric was collected.
// In an ideal world, the "working set" is the amount of memory in-use that cannot be freed under memory pressure.
// However, calculation of the working set varies by host OS, and generally makes heavy
// use of heuristics to produce an estimate.
// The Kubernetes model for a container's working set expects that the container runtime counts anonymous
// memory associated with the container in question. The working set metric typically also includes
// some cached (file-backed) memory, because the host OS cannot always reclaim pages.
func getPodMetrics(rawMetrics []metricsapi.PodMetrics, resource v1.ResourceName) PodMetricsInfo {
	res := make(PodMetricsInfo, len(rawMetrics))
	for _, m := range rawMetrics {
		podSum := int64(0)
		missing := len(m.Containers) == 0
		for _, container := range m.Containers {
			resValue, found := container.Usage[resource]
			if !found {
				missing = true
				logger.Info(fmt.Sprintf("missing resource metric, pod=(%s/%s), resourceMetric=%s", m.Namespace, m.Name, resource))
				break
			}
			podSum += resValue.MilliValue()
		}
		if !missing {
			res[client.ObjectKey{Namespace: m.Namespace, Name: m.Name}] = PodMetric{
				Timestamp: m.Timestamp.Time,
				Window:    m.Window.Duration,
				Value:     podSum,
			}
		}
	}
	return res
}