Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add Aggregation Storage #1213

Merged
merged 10 commits into from
Feb 16, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 1 addition & 7 deletions sdk/include/opentelemetry/sdk/metrics/data/point_data.h
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
#ifndef ENABLE_METRICS_PREVIEW
# include "opentelemetry/common/timestamp.h"
# include "opentelemetry/nostd/variant.h"
# include "opentelemetry/sdk/metrics/instruments.h"
# include "opentelemetry/version.h"

# include <vector>
Expand All @@ -18,13 +19,6 @@ namespace metrics
using ValueType = nostd::variant<long, double>;
using ListType = nostd::variant<std::vector<long>, std::vector<double>>;

enum class AggregationTemporarily
{
kUnspecified,
kDelta,
kCummulative
};

class SumPointData
{
public:
Expand Down
27 changes: 26 additions & 1 deletion sdk/include/opentelemetry/sdk/metrics/instruments.h
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@

#pragma once
#ifndef ENABLE_METRICS_PREVIEW
# include "opentelemetry/nostd/string_view.h"
# include "opentelemetry/sdk/common/attribute_utils.h"
OPENTELEMETRY_BEGIN_NAMESPACE
namespace sdk
{
Expand All @@ -27,6 +27,22 @@ enum class InstrumentValueType
kDouble
};

enum class AggregationType
{
kDrop,
kHistogram,
kLastValue,
kSum,
kDefault
};

enum class AggregationTemporarily
{
kUnspecified,
kDelta,
kCummulative
};

struct InstrumentDescriptor
{
std::string name_;
Expand All @@ -36,6 +52,15 @@ struct InstrumentDescriptor
InstrumentValueType value_type_;
};

using MetricAttributes = opentelemetry::sdk::common::OrderedAttributeMap;

// TBD -> Remove once MetricCollector is imoplemeted
class MetricCollector
{
public:
AggregationTemporarily aggregation_temporarily_;
};

/*class InstrumentSelector {
public:
InstrumentSelector(opentelemetry::nostd::string_view name,
Expand Down
16 changes: 12 additions & 4 deletions sdk/include/opentelemetry/sdk/metrics/measurement_processor.h
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ static std::size_t MakeKey(const MetricReader &metric_reader)
{
return reinterpret_cast<std::size_t>(&metric_reader);
}

class MeasurementProcessor
{
public:
Expand All @@ -41,12 +42,15 @@ class MeasurementProcessor

class DefaultMeasurementProcessor : public MeasurementProcessor
{

public:
bool AddMetricStorage(const MetricReader &reader)
{
// TBD = check if already present.
metric_storages_[MakeKey(reader)] = std::unique_ptr<SyncMetricStorage>(new SyncMetricStorage());
// TBD Check if already present
// pass intrumentation type, and aggregation type instead of hardcodig below.
InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kCounter,
InstrumentValueType::kLong};
metric_storages_[MakeKey(reader)] = std::unique_ptr<SyncMetricStorage>(
new SyncMetricStorage(instr_desc, AggregationType::kSum));
return true;
}

Expand Down Expand Up @@ -91,9 +95,13 @@ class DefaultMeasurementProcessor : public MeasurementProcessor
nostd::function_ref<bool(MetricData)> callback) noexcept override
{
auto i = metric_storages_.find(MakeKey(reader));

// TBD - Remove hardcodings below
std::vector<MetricCollector *> collectors;
if (i != metric_storages_.end())
{
return i->second->Collect(aggregation_temporarily, callback);

return i->second->Collect(nullptr, collectors, nullptr, nullptr, callback);
}
return false;
}
Expand Down
127 changes: 127 additions & 0 deletions sdk/include/opentelemetry/sdk/metrics/state/attributes_hashmap.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

#pragma once
#ifndef ENABLE_METRICS_PREVIEW
# include "opentelemetry/common/spin_lock_mutex.h"
# include "opentelemetry/nostd/function_ref.h"
# include "opentelemetry/sdk/common/attribute_utils.h"
# include "opentelemetry/sdk/common/attributemap_hash.h"
# include "opentelemetry/sdk/metrics/aggregation/aggregation.h"
# include "opentelemetry/sdk/metrics/instruments.h"
# include "opentelemetry/version.h"

# include <functional>
# include <memory>
# include <mutex>
# include <unordered_map>

OPENTELEMETRY_BEGIN_NAMESPACE
namespace sdk
{
namespace metrics
{
using opentelemetry::sdk::common::OrderedAttributeMap;

class AttributeHashGenerator
{
public:
size_t operator()(const MetricAttributes &attributes) const
{
return opentelemetry::sdk::common::GetHashForAttributeMap(attributes);
}
};

class AttributesHashMap
{
public:
Aggregation *Get(const MetricAttributes &attributes) const
{
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
auto it = hash_map_.find(attributes);
if (it != hash_map_.end())
{
return it->second.get();
}
return nullptr;
}

/**
* @return check if key is present in hash
*
*/
bool Has(const MetricAttributes &attributes) const
{
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
return (hash_map_.find(attributes) == hash_map_.end()) ? false : true;
}

/**
* @return the pointer to value for given key if present.
* If not present, it uses the provided callback to generate
* value and store in the hash
*/
Aggregation *GetOrSetDefault(const MetricAttributes &attributes,
std::function<std::unique_ptr<Aggregation>()> aggregation_callback)
{
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
auto it = hash_map_.find(attributes);
if (it != hash_map_.end())
{
return it->second.get();
}

hash_map_[attributes] = std::move(aggregation_callback());
return hash_map_[attributes].get();
}

/**
* Set the value for given key, overwriting the value if already present
*/
void Set(const MetricAttributes &attributes, std::unique_ptr<Aggregation> value)
esigo marked this conversation as resolved.
Show resolved Hide resolved
{
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
hash_map_[attributes] = std::move(value);
}

/**
* Iterate the hash to yield key and value stored in hash.
*/
bool GetAllEnteries(
nostd::function_ref<bool(const MetricAttributes &, Aggregation &)> callback) const
{
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
for (auto &kv : hash_map_)
{
if (!callback(kv.first, *(kv.second.get())))
{
return false; // callback is not prepared to consume data
}
}
return true;
}

/**
* Return the size of hash.
*/
size_t Size()
{
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(GetLock());
return hash_map_.size();
}

private:
std::unordered_map<MetricAttributes, std::unique_ptr<Aggregation>, AttributeHashGenerator>
hash_map_;

static opentelemetry::common::SpinLockMutex &GetLock() noexcept
{
static opentelemetry::common::SpinLockMutex lock;
return lock;
}
};
} // namespace metrics

} // namespace sdk
OPENTELEMETRY_END_NAMESPACE
#endif
16 changes: 12 additions & 4 deletions sdk/include/opentelemetry/sdk/metrics/state/metric_storage.h
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,12 @@ class MetricStorage
{
public:
/* collect the metrics from this storage */
virtual bool Collect(AggregationTemporarily aggregation_temporarily,
nostd::function_ref<bool(MetricData)> callback) noexcept = 0;
virtual bool Collect(
MetricCollector *collector,
nostd::span<MetricCollector *> collectors,
opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary *instrumentation_library,
opentelemetry::sdk::resource::Resource *resource,
nostd::function_ref<bool(MetricData)> callback) noexcept = 0;
};

class WritableMetricStorage
Expand All @@ -39,8 +43,12 @@ class WritableMetricStorage
class NoopMetricStorage : public MetricStorage
{
public:
bool Collect(AggregationTemporarily aggregation_temporarily,
nostd::function_ref<bool(MetricData)> callback) noexcept override
bool Collect(
MetricCollector *collector,
nostd::span<MetricCollector *> collectors,
opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary *instrumentation_library,
opentelemetry::sdk::resource::Resource *resource,
nostd::function_ref<bool(MetricData)> callback) noexcept override
{
MetricData metric_data;
if (callback(metric_data))
Expand Down
Loading