// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you 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.

#include "util/metrics.h"

#include <glog/logging.h>

#include <rapidjson/encodings.h>

#include <rapidjson/stringbuffer.h>

#include <rapidjson/writer.h>

#include <initializer_list>

#include "common/config.h"

namespace doris {

std::ostream& operator<<(std::ostream& os, MetricType type) {

    switch (type) {

    case MetricType::COUNTER:

        os << "counter";

        break;

    case MetricType::GAUGE:

        os << "gauge";

        break;

    case MetricType::HISTOGRAM:

        os << "histogram";

        break;

    case MetricType::SUMMARY:

        os << "summary";

        break;

    case MetricType::UNTYPED:

        os << "untyped";

        break;

    default:

        os << "unknown";

        break;

    }

    return os;

}

const char* unit_name(MetricUnit unit) {

    switch (unit) {

    case MetricUnit::NANOSECONDS:

        return "nanoseconds";

    case MetricUnit::MICROSECONDS:

        return "microseconds";

    case MetricUnit::MILLISECONDS:

        return "milliseconds";

    case MetricUnit::SECONDS:

        return "seconds";

    case MetricUnit::BYTES:

        return "bytes";

    case MetricUnit::ROWS:

        return "rows";

    case MetricUnit::PERCENT:

        return "percent";

    case MetricUnit::REQUESTS:

        return "requests";

    case MetricUnit::OPERATIONS:

        return "operations";

    case MetricUnit::BLOCKS:

        return "blocks";

    case MetricUnit::ROWSETS:

        return "rowsets";

    case MetricUnit::CONNECTIONS:

        return "rowsets";

    default:

        return "nounit";

    }

}

std::string labels_to_string(std::initializer_list<const Labels*> multi_labels) {

    bool all_empty = true;

    for (const auto& labels : multi_labels) {

        if (!labels->empty()) {

            all_empty = false;

            break;

        }

    }

    if (all_empty) {

        return std::string();

    }

    std::stringstream ss;

    ss << "{";

    int i = 0;

    for (auto labels : multi_labels) {

        for (const auto& label : *labels) {

            if (i++ > 0) {

                ss << ",";

            }

            ss << label.first << "=\"" << label.second << "\"";

        }

    }

    ss << "}";

    return ss.str();

}

std::string Metric::to_prometheus(const std::string& display_name, const Labels& entity_labels,

                                  const Labels& metric_labels) const {

    std::stringstream ss;

    ss << display_name                                       // metric name

       << labels_to_string({&entity_labels, &metric_labels}) // metric labels

       << " " << to_string() << "\n";                        // metric value

    return ss.str();

}

std::map<std::string, double> HistogramMetric::_s_output_percentiles = {

        {"0.50", 50.0}, {"0.75", 75.0}, {"0.90", 90.0}, {"0.95", 95.0}, {"0.99", 99.0}};

void HistogramMetric::clear() {

    std::lock_guard<std::mutex> l(_lock);

    _stats.clear();

}

bool HistogramMetric::is_empty() const {

    return _stats.is_empty();

}

void HistogramMetric::add(const uint64_t& value) {

    _stats.add(value);

}

void HistogramMetric::merge(const HistogramMetric& other) {

    std::lock_guard<std::mutex> l(_lock);

    _stats.merge(other._stats);

}

void HistogramMetric::set_histogram(const HistogramStat& stats) {

    std::lock_guard<std::mutex> l(_lock);

    _stats.clear();

    _stats.merge(stats);

}

double HistogramMetric::median() const {

    return _stats.median();

}

double HistogramMetric::percentile(double p) const {

    return _stats.percentile(p);

}

double HistogramMetric::average() const {

    return _stats.average();

}

double HistogramMetric::standard_deviation() const {

    return _stats.standard_deviation();

}

std::string HistogramMetric::to_string() const {

    return _stats.to_string();

}

std::string HistogramMetric::to_prometheus(const std::string& display_name,

                                           const Labels& entity_labels,

                                           const Labels& metric_labels) const {

    std::stringstream ss;

    for (const auto& percentile : _s_output_percentiles) {

        auto quantile_lable = Labels({{"quantile", percentile.first}});

        ss << display_name << labels_to_string({&entity_labels, &metric_labels, &quantile_lable})

           << " " << _stats.percentile(percentile.second) << "\n";

    }

    ss << display_name << "_sum" << labels_to_string({&entity_labels, &metric_labels}) << " "

       << _stats.sum() << "\n";

    ss << display_name << "_count" << labels_to_string({&entity_labels, &metric_labels}) << " "

       << _stats.num() << "\n";

    ss << display_name << "_max" << labels_to_string({&entity_labels, &metric_labels}) << " "

       << _stats.max() << "\n";

    ss << display_name << "_min" << labels_to_string({&entity_labels, &metric_labels}) << " "

       << _stats.min() << "\n";

    ss << display_name << "_average" << labels_to_string({&entity_labels, &metric_labels}) << " "

       << _stats.average() << "\n";

    ss << display_name << "_median" << labels_to_string({&entity_labels, &metric_labels}) << " "

       << _stats.median() << "\n";

    ss << display_name << "_standard_deviation"

       << labels_to_string({&entity_labels, &metric_labels}) << " " << _stats.standard_deviation()

       << "\n";

    return ss.str();

}

rj::Value HistogramMetric::to_json_value(rj::Document::AllocatorType& allocator) const {

    rj::Value json_value(rj::kObjectType);

    json_value.AddMember("total_count", rj::Value(_stats.num()), allocator);

    json_value.AddMember("min", rj::Value(_stats.min()), allocator);

    json_value.AddMember("average", rj::Value(_stats.average()), allocator);

    json_value.AddMember("median", rj::Value(_stats.median()), allocator);

    for (const auto& percentile : _s_output_percentiles) {

        json_value.AddMember(

                rj::Value(std::string("percentile_").append(percentile.first.substr(2)).c_str(),

                          allocator),

                rj::Value(_stats.percentile(percentile.second)), allocator);

    }

    json_value.AddMember("standard_deviation", rj::Value(_stats.standard_deviation()), allocator);

    json_value.AddMember("max", rj::Value(_stats.max()), allocator);

    json_value.AddMember("total_sum", rj::Value(_stats.sum()), allocator);

    return json_value;

}

std::string MetricPrototype::simple_name() const {

    return group_name.empty() ? name : group_name;

}

std::string MetricPrototype::combine_name(const std::string& registry_name) const {

    return (registry_name.empty() ? std::string() : registry_name + "_") + simple_name();

}

std::string MetricPrototype::to_prometheus(const std::string& registry_name) const {

    std::stringstream ss;

    ss << "# TYPE " << combine_name(registry_name) << " " << type << "\n";

    return ss.str();

}

void MetricEntity::deregister_metric(const MetricPrototype* metric_type) {

    std::lock_guard<std::mutex> l(_lock);

    auto metric = _metrics.find(metric_type);

    if (metric != _metrics.end()) {

        delete metric->second;

        _metrics.erase(metric);

    }

}

Metric* MetricEntity::get_metric(const std::string& name, const std::string& group_name) const {

    MetricPrototype dummy(MetricType::UNTYPED, MetricUnit::NOUNIT, name, "", group_name);

    std::lock_guard<std::mutex> l(_lock);

    auto it = _metrics.find(&dummy);

    if (it == _metrics.end()) {

        return nullptr;

    }

    return it->second;

}

void MetricEntity::register_hook(const std::string& name, const std::function<void()>& hook) {

    std::lock_guard<std::mutex> l(_lock);

#ifndef BE_TEST

    DCHECK(_hooks.find(name) == _hooks.end()) << "hook is already exist! " << _name << ":" << name;

#endif

    _hooks.emplace(name, hook);

}

void MetricEntity::deregister_hook(const std::string& name) {

    std::lock_guard<std::mutex> l(_lock);

    _hooks.erase(name);

}

void MetricEntity::trigger_hook_unlocked(bool force) const {

    // When 'enable_metric_calculator' is true, hooks will be triggered by a background thread,

    // see 'calculate_metrics' in daemon.cpp for more details.

    if (!force && config::enable_metric_calculator) {

        return;

    }

    for (const auto& hook : _hooks) {

        hook.second();

    }

}

MetricRegistry::~MetricRegistry() {}

std::shared_ptr<MetricEntity> MetricRegistry::register_entity(const std::string& name,

                                                              const Labels& labels,

                                                              MetricEntityType type) {

    std::shared_ptr<MetricEntity> entity = std::make_shared<MetricEntity>(type, name, labels);

    std::lock_guard<std::mutex> l(_lock);

    auto inserted_entity = _entities.insert(std::make_pair(entity, 1));

    if (!inserted_entity.second) {

        // If exist, increase the registered count

        inserted_entity.first->second++;

    }

    return inserted_entity.first->first;

}

void MetricRegistry::deregister_entity(const std::shared_ptr<MetricEntity>& entity) {

    std::lock_guard<std::mutex> l(_lock);

    auto found_entity = _entities.find(entity);

    if (found_entity != _entities.end()) {

        // Decrease the registered count

        --found_entity->second;

        if (found_entity->second == 0) {

            // Only erase it when registered count is zero

            _entities.erase(found_entity);

        }

    }

}

std::shared_ptr<MetricEntity> MetricRegistry::get_entity(const std::string& name,

                                                         const Labels& labels,

                                                         MetricEntityType type) {

    std::shared_ptr<MetricEntity> dummy = std::make_shared<MetricEntity>(type, name, labels);

    std::lock_guard<std::mutex> l(_lock);

    auto entity = _entities.find(dummy);

    if (entity == _entities.end()) {

        return std::shared_ptr<MetricEntity>();

    }

    return entity->first;

}

void MetricRegistry::trigger_all_hooks(bool force) const {

    std::lock_guard<std::mutex> l(_lock);

    for (const auto& entity : _entities) {

        std::lock_guard<std::mutex> l(entity.first->_lock);

        entity.first->trigger_hook_unlocked(force);

    }

}

std::string MetricRegistry::to_prometheus(bool with_tablet_metrics) const {

    // Reorder by MetricPrototype

    EntityMetricsByType entity_metrics_by_types;

    std::lock_guard<std::mutex> l(_lock);

    for (const auto& entity : _entities) {

        if (entity.first->_type == MetricEntityType::kTablet && !with_tablet_metrics) {

            continue;

        }

        std::lock_guard<std::mutex> l(entity.first->_lock);

        entity.first->trigger_hook_unlocked(false);

        for (const auto& metric : entity.first->_metrics) {

            std::pair<MetricEntity*, Metric*> new_elem =

                    std::make_pair(entity.first.get(), metric.second);

            auto found = entity_metrics_by_types.find(metric.first);

            if (found == entity_metrics_by_types.end()) {

                entity_metrics_by_types.emplace(

                        metric.first, std::vector<std::pair<MetricEntity*, Metric*>>({new_elem}));

            } else {

                found->second.emplace_back(new_elem);

            }

        }

    }

    // Output

    std::stringstream ss;

    std::string last_group_name;

    for (const auto& entity_metrics_by_type : entity_metrics_by_types) {

        if (last_group_name.empty() ||

            last_group_name != entity_metrics_by_type.first->group_name) {

            ss << entity_metrics_by_type.first->to_prometheus(_name); // metric TYPE line

        }

        last_group_name = entity_metrics_by_type.first->group_name;

        std::string display_name = entity_metrics_by_type.first->combine_name(_name);

        for (const auto& entity_metric : entity_metrics_by_type.second) {

            ss << entity_metric.second->to_prometheus(display_name, // metric key-value line

                                                      entity_metric.first->_labels,

                                                      entity_metrics_by_type.first->labels);

        }

    }

    return ss.str();

}

std::string MetricRegistry::to_json(bool with_tablet_metrics) const {

    rj::Document doc {rj::kArrayType};

    rj::Document::AllocatorType& allocator = doc.GetAllocator();

    std::lock_guard<std::mutex> l(_lock);

    for (const auto& entity : _entities) {

        if (entity.first->_type == MetricEntityType::kTablet && !with_tablet_metrics) {

            continue;

        }

        std::lock_guard<std::mutex> l(entity.first->_lock);

        entity.first->trigger_hook_unlocked(false);

        for (const auto& metric : entity.first->_metrics) {

            rj::Value metric_obj(rj::kObjectType);

            // tags

            rj::Value tag_obj(rj::kObjectType);

            tag_obj.AddMember("metric", rj::Value(metric.first->simple_name().c_str(), allocator),

                              allocator);

            // MetricPrototype's labels

            for (auto& label : metric.first->labels) {

                tag_obj.AddMember(rj::Value(label.first.c_str(), allocator),

                                  rj::Value(label.second.c_str(), allocator), allocator);

            }

            // MetricEntity's labels

            for (auto& label : entity.first->_labels) {

                tag_obj.AddMember(rj::Value(label.first.c_str(), allocator),

                                  rj::Value(label.second.c_str(), allocator), allocator);

            }

            metric_obj.AddMember("tags", tag_obj, allocator);

            // unit

            rj::Value unit_val(unit_name(metric.first->unit), allocator);

            metric_obj.AddMember("unit", unit_val, allocator);

            // value

            metric_obj.AddMember("value", metric.second->to_json_value(allocator), allocator);

            doc.PushBack(metric_obj, allocator);

        }

    }

    rj::StringBuffer strBuf;

    rj::Writer<rj::StringBuffer> writer(strBuf);

    doc.Accept(writer);

    return strBuf.GetString();

}

std::string MetricRegistry::to_core_string() const {

    std::stringstream ss;

    std::lock_guard<std::mutex> l(_lock);

    for (const auto& entity : _entities) {

        std::lock_guard<std::mutex> l(entity.first->_lock);

        entity.first->trigger_hook_unlocked(false);

        for (const auto& metric : entity.first->_metrics) {

            if (metric.first->is_core_metric) {

                ss << metric.first->combine_name(_name) << " LONG " << metric.second->to_string()

                   << "\n";

            }

        }

    }

    return ss.str();

}

} // namespace doris