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

#pragma once

#include <gen_cpp/Metrics_types.h>

#include <atomic>

#include <cstdint>

#include <memory>

#include <mutex>

#include <unordered_map>

#include "common/metrics/doris_metrics.h"

#include "common/metrics/metrics.h"

#include "io/io_common.h"

#include "runtime/runtime_profile.h"

#include "storage/olap_common.h"

namespace doris {

namespace io {

struct AtomicStatistics {

    std::atomic<int64_t> num_io_bytes_read_from_cache = 0;

    std::atomic<int64_t> num_io_bytes_read_from_remote = 0;

    std::atomic<int64_t> num_io_bytes_read_from_peer = 0;

    std::atomic<int64_t> inverted_index_bytes_read_from_remote = 0;

    std::atomic<int64_t> segment_footer_index_bytes_read_from_remote = 0;

};

class FileCacheMetrics {

public:

    static FileCacheMetrics& instance() {

        static FileCacheMetrics s_metrics;

        return s_metrics;

    }

    FileCacheMetrics() {

        FileCacheStatistics stats;

        update(&stats);

    }

    void update(FileCacheStatistics* stats);

    std::shared_ptr<AtomicStatistics> report();

private:

    void register_entity();

    void update_metrics_callback();

private:

    std::mutex _mtx;

    // use shared_ptr for concurrent

    std::shared_ptr<AtomicStatistics> _statistics;

};

FileCacheStatistics diff_file_cache_statistics(const FileCacheStatistics& current,

                                               const FileCacheStatistics& previous);

struct FileCacheProfileReporter {

    RuntimeProfile* _profile = nullptr;

    RuntimeProfile::Counter* num_local_io_total = nullptr;

    RuntimeProfile::Counter* num_remote_io_total = nullptr;

    RuntimeProfile::Counter* num_peer_io_total = nullptr;

    RuntimeProfile::Counter* local_io_timer = nullptr;

    RuntimeProfile::Counter* bytes_scanned_from_cache = nullptr;

    RuntimeProfile::Counter* bytes_scanned_from_remote = nullptr;

    RuntimeProfile::Counter* bytes_scanned_from_peer = nullptr;

    RuntimeProfile::Counter* remote_io_timer = nullptr;

    RuntimeProfile::Counter* peer_io_timer = nullptr;

    RuntimeProfile::Counter* remote_wait_timer = nullptr;

    RuntimeProfile::Counter* write_cache_io_timer = nullptr;

    RuntimeProfile::Counter* bytes_write_into_cache = nullptr;

    RuntimeProfile::Counter* num_skip_cache_io_total = nullptr;

    RuntimeProfile::Counter* read_cache_file_directly_timer = nullptr;

    RuntimeProfile::Counter* cache_get_or_set_timer = nullptr;

    RuntimeProfile::Counter* lock_wait_timer = nullptr;

    RuntimeProfile::Counter* get_timer = nullptr;

    RuntimeProfile::Counter* set_timer = nullptr;

    RuntimeProfile::Counter* inverted_index_num_local_io_total = nullptr;

    RuntimeProfile::Counter* inverted_index_num_remote_io_total = nullptr;

    RuntimeProfile::Counter* inverted_index_num_peer_io_total = nullptr;

    RuntimeProfile::Counter* inverted_index_bytes_scanned_from_cache = nullptr;

    RuntimeProfile::Counter* inverted_index_bytes_scanned_from_remote = nullptr;

    RuntimeProfile::Counter* inverted_index_bytes_scanned_from_peer = nullptr;

    RuntimeProfile::Counter* inverted_index_local_io_timer = nullptr;

    RuntimeProfile::Counter* inverted_index_remote_io_timer = nullptr;

    RuntimeProfile::Counter* inverted_index_peer_io_timer = nullptr;

    RuntimeProfile::Counter* inverted_index_io_timer = nullptr;

    RuntimeProfile::Counter* segment_footer_index_num_local_io_total = nullptr;

    RuntimeProfile::Counter* segment_footer_index_num_remote_io_total = nullptr;

    RuntimeProfile::Counter* segment_footer_index_num_peer_io_total = nullptr;

    RuntimeProfile::Counter* segment_footer_index_bytes_scanned_from_cache = nullptr;

    RuntimeProfile::Counter* segment_footer_index_bytes_scanned_from_remote = nullptr;

    RuntimeProfile::Counter* segment_footer_index_bytes_scanned_from_peer = nullptr;

    RuntimeProfile::Counter* segment_footer_index_local_io_timer = nullptr;

    RuntimeProfile::Counter* segment_footer_index_remote_io_timer = nullptr;

    RuntimeProfile::Counter* segment_footer_index_peer_io_timer = nullptr;

    // Cross-CG / Same-CG peer read counters

    RuntimeProfile::Counter* num_cross_cg_peer_io_total = nullptr;

    RuntimeProfile::Counter* bytes_scanned_from_cross_cg_peer = nullptr;

    RuntimeProfile::Counter* cross_cg_peer_io_timer = nullptr;

    RuntimeProfile::Counter* num_same_cg_peer_io_total = nullptr;

    RuntimeProfile::Counter* bytes_scanned_from_same_cg_peer = nullptr;

    RuntimeProfile::Counter* same_cg_peer_io_timer = nullptr;

    RuntimeProfile::Counter* num_peer_race_peer_win = nullptr;

    RuntimeProfile::Counter* num_peer_race_s3_win = nullptr;

    RuntimeProfile::Counter* num_peer_lazy_fetch = nullptr;

    RuntimeProfile::Counter* peer_lazy_fetch_timer = nullptr;

    FileCacheProfileReporter(RuntimeProfile* profile);

    void update(const FileCacheStatistics* statistics) const;

};

} // namespace io

} // namespace doris