blob: 70d69dc14a69090e1c91304aef2b889e1e552428 [file] [log] [blame]
// 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 <stdlib.h>
#include <algorithm>
#include <iostream>
#include <boost/bind.hpp>
#include "common/object-pool.h"
#include "testutil/gtest-util.h"
#include "util/container-util.h"
#include "util/periodic-counter-updater.h"
#include "util/runtime-profile-counters.h"
#include "util/thread.h"
#include "common/names.h"
DECLARE_int32(status_report_interval_ms);
DECLARE_int32(periodic_counter_update_period_ms);
namespace impala {
TEST(CountersTest, Basic) {
ObjectPool pool;
RuntimeProfile* profile_a = RuntimeProfile::Create(&pool, "ProfileA");
RuntimeProfile* profile_a1 = RuntimeProfile::Create(&pool, "ProfileA1");
RuntimeProfile* profile_a2 = RuntimeProfile::Create(&pool, "ProfileAb");
TRuntimeProfileTree thrift_profile;
profile_a->AddChild(profile_a1);
profile_a->AddChild(profile_a2);
// Test Empty
profile_a->ToThrift(&thrift_profile);
EXPECT_EQ(thrift_profile.nodes.size(), 3);
thrift_profile.nodes.clear();
RuntimeProfile::Counter* counter_a;
RuntimeProfile::Counter* counter_b;
RuntimeProfile::Counter* counter_merged;
// Updating/setting counter
counter_a = profile_a->AddCounter("A", TUnit::UNIT);
EXPECT_TRUE(counter_a != NULL);
counter_a->Add(10);
counter_a->Add(-5);
EXPECT_EQ(counter_a->value(), 5);
counter_a->Set(1);
EXPECT_EQ(counter_a->value(), 1);
counter_b = profile_a2->AddCounter("B", TUnit::BYTES);
EXPECT_TRUE(counter_b != NULL);
// Update status to be included in ExecSummary
TExecSummary exec_summary;
TStatus status;
status.__set_status_code(TErrorCode::CANCELLED);
exec_summary.__set_status(status);
profile_a->SetTExecSummary(exec_summary);
// Serialize/deserialize to thrift
profile_a->ToThrift(&thrift_profile);
RuntimeProfile* from_thrift = RuntimeProfile::CreateFromThrift(&pool, thrift_profile);
counter_merged = from_thrift->GetCounter("A");
EXPECT_EQ(counter_merged->value(), 1);
EXPECT_TRUE(from_thrift->GetCounter("Not there") == NULL);
TExecSummary exec_summary_result;
from_thrift->GetExecSummary(&exec_summary_result);
EXPECT_EQ(exec_summary_result.status, status);
// Serialize/deserialize to archive string
string archive_str;
EXPECT_OK(profile_a->SerializeToArchiveString(&archive_str));
TRuntimeProfileTree deserialized_thrift_profile;
EXPECT_OK(RuntimeProfile::DeserializeFromArchiveString(
archive_str, &deserialized_thrift_profile));
RuntimeProfile* deserialized_profile =
RuntimeProfile::CreateFromThrift(&pool, deserialized_thrift_profile);
counter_merged = deserialized_profile->GetCounter("A");
EXPECT_EQ(counter_merged->value(), 1);
EXPECT_TRUE(deserialized_profile->GetCounter("Not there") == nullptr);
deserialized_profile->GetExecSummary(&exec_summary_result);
EXPECT_EQ(exec_summary_result.status, status);
// Averaged
RuntimeProfile* averaged_profile = RuntimeProfile::Create(&pool, "Merged", true);
averaged_profile->UpdateAverage(from_thrift);
counter_merged = averaged_profile->GetCounter("A");
EXPECT_EQ(counter_merged->value(), 1);
// UpdateAverage again, there should be no change.
averaged_profile->UpdateAverage(from_thrift);
EXPECT_EQ(counter_merged->value(), 1);
counter_a = profile_a2->AddCounter("A", TUnit::UNIT);
counter_a->Set(3);
averaged_profile->UpdateAverage(profile_a2);
EXPECT_EQ(counter_merged->value(), 2);
// Update
RuntimeProfile* updated_profile = RuntimeProfile::Create(&pool, "Updated");
updated_profile->Update(thrift_profile);
RuntimeProfile::Counter* counter_updated = updated_profile->GetCounter("A");
EXPECT_EQ(counter_updated->value(), 1);
// Update 2 more times, counters should stay the same
updated_profile->Update(thrift_profile);
updated_profile->Update(thrift_profile);
EXPECT_EQ(counter_updated->value(), 1);
}
void ValidateCounter(RuntimeProfile* profile, const string& name, int64_t value) {
RuntimeProfile::Counter* counter = profile->GetCounter(name);
EXPECT_TRUE(counter != NULL);
EXPECT_EQ(counter->value(), value);
}
TEST(CountersTest, MergeAndUpdate) {
// Create two trees. Each tree has two children, one of which has the
// same name in both trees. Merging the two trees should result in 3
// children, with the counters from the shared child aggregated.
ObjectPool pool;
RuntimeProfile* profile1 = RuntimeProfile::Create(&pool, "Parent1");
RuntimeProfile* p1_child1 = RuntimeProfile::Create(&pool, "Child1");
RuntimeProfile* p1_child2 = RuntimeProfile::Create(&pool, "Child2");
profile1->AddChild(p1_child1);
profile1->AddChild(p1_child2);
RuntimeProfile* profile2 = RuntimeProfile::Create(&pool, "Parent2");
RuntimeProfile* p2_child1 = RuntimeProfile::Create(&pool, "Child1");
RuntimeProfile* p2_child3 = RuntimeProfile::Create(&pool, "Child3");
profile2->AddChild(p2_child1);
profile2->AddChild(p2_child3);
// Create parent level counters
RuntimeProfile::Counter* parent1_shared =
profile1->AddCounter("Parent Shared", TUnit::UNIT);
RuntimeProfile::Counter* parent2_shared =
profile2->AddCounter("Parent Shared", TUnit::UNIT);
RuntimeProfile::Counter* parent1_only =
profile1->AddCounter("Parent 1 Only", TUnit::UNIT);
RuntimeProfile::Counter* parent2_only =
profile2->AddCounter("Parent 2 Only", TUnit::UNIT);
parent1_shared->Add(1);
parent2_shared->Add(3);
parent1_only->Add(2);
parent2_only->Add(5);
// Create child level counters
RuntimeProfile::Counter* p1_c1_shared =
p1_child1->AddCounter("Child1 Shared", TUnit::UNIT);
RuntimeProfile::Counter* p1_c1_only =
p1_child1->AddCounter("Child1 Parent 1 Only", TUnit::UNIT);
RuntimeProfile::Counter* p1_c2 =
p1_child2->AddCounter("Child2", TUnit::UNIT);
RuntimeProfile::Counter* p2_c1_shared =
p2_child1->AddCounter("Child1 Shared", TUnit::UNIT);
RuntimeProfile::Counter* p2_c1_only =
p1_child1->AddCounter("Child1 Parent 2 Only", TUnit::UNIT);
RuntimeProfile::Counter* p2_c3 =
p2_child3->AddCounter("Child3", TUnit::UNIT);
p1_c1_shared->Add(10);
p1_c1_only->Add(50);
p2_c1_shared->Add(20);
p2_c1_only->Add(100);
p2_c3->Add(30);
p1_c2->Add(40);
// Merge the two and validate
TRuntimeProfileTree tprofile1;
profile1->ToThrift(&tprofile1);
RuntimeProfile* averaged_profile = RuntimeProfile::Create(&pool, "merged", true);
averaged_profile->UpdateAverage(profile1);
averaged_profile->UpdateAverage(profile2);
EXPECT_EQ(5, averaged_profile->num_counters());
ValidateCounter(averaged_profile, "Parent Shared", 2);
ValidateCounter(averaged_profile, "Parent 1 Only", 2);
ValidateCounter(averaged_profile, "Parent 2 Only", 5);
vector<RuntimeProfile*> children;
averaged_profile->GetChildren(&children);
EXPECT_EQ(children.size(), 3);
for (int i = 0; i < 3; ++i) {
RuntimeProfile* profile = children[i];
if (profile->name().compare("Child1") == 0) {
EXPECT_EQ(5, profile->num_counters());
ValidateCounter(profile, "Child1 Shared", 15);
ValidateCounter(profile, "Child1 Parent 1 Only", 50);
ValidateCounter(profile, "Child1 Parent 2 Only", 100);
} else if (profile->name().compare("Child2") == 0) {
EXPECT_EQ(3, profile->num_counters());
ValidateCounter(profile, "Child2", 40);
} else if (profile->name().compare("Child3") == 0) {
EXPECT_EQ(3, profile->num_counters());
ValidateCounter(profile, "Child3", 30);
} else {
FAIL();
}
}
// make sure we can print
stringstream dummy;
averaged_profile->PrettyPrint(&dummy);
// Update profile2 w/ profile1 and validate
profile2->Update(tprofile1);
EXPECT_EQ(5, profile2->num_counters());
ValidateCounter(profile2, "Parent Shared", 1);
ValidateCounter(profile2, "Parent 1 Only", 2);
ValidateCounter(profile2, "Parent 2 Only", 5);
profile2->GetChildren(&children);
EXPECT_EQ(children.size(), 3);
for (int i = 0; i < 3; ++i) {
RuntimeProfile* profile = children[i];
if (profile->name().compare("Child1") == 0) {
EXPECT_EQ(5, profile->num_counters());
ValidateCounter(profile, "Child1 Shared", 10);
ValidateCounter(profile, "Child1 Parent 1 Only", 50);
ValidateCounter(profile, "Child1 Parent 2 Only", 100);
} else if (profile->name().compare("Child2") == 0) {
EXPECT_EQ(3, profile->num_counters());
ValidateCounter(profile, "Child2", 40);
} else if (profile->name().compare("Child3") == 0) {
EXPECT_EQ(3, profile->num_counters());
ValidateCounter(profile, "Child3", 30);
} else {
FAIL();
}
}
// make sure we can print
profile2->PrettyPrint(&dummy);
}
// Regression test for IMPALA-6694 - child order isn't preserved if a child
// is prepended between updates.
TEST(CountersTest, MergeAndUpdateChildOrder) {
ObjectPool pool;
// Add Child2 first.
RuntimeProfile* profile1 = RuntimeProfile::Create(&pool, "Parent");
RuntimeProfile* p1_child2 = RuntimeProfile::Create(&pool, "Child2");
profile1->AddChild(p1_child2);
TRuntimeProfileTree tprofile1_v1, tprofile1_v2, tprofile1_v3;
profile1->ToThrift(&tprofile1_v1);
// Update averaged and deserialized profiles from the serialized profile.
RuntimeProfile* averaged_profile = RuntimeProfile::Create(&pool, "merged", true);
RuntimeProfile* deserialized_profile = RuntimeProfile::Create(&pool, "Parent");
averaged_profile->UpdateAverage(profile1);
deserialized_profile->Update(tprofile1_v1);
std::vector<RuntimeProfile*> tmp_children;
averaged_profile->GetChildren(&tmp_children);
EXPECT_EQ(1, tmp_children.size());
EXPECT_EQ("Child2", tmp_children[0]->name());
deserialized_profile->GetChildren(&tmp_children);
EXPECT_EQ(1, tmp_children.size());
EXPECT_EQ("Child2", tmp_children[0]->name());
// Prepend Child1 and update profiles.
RuntimeProfile* p1_child1 = RuntimeProfile::Create(&pool, "Child1");
profile1->PrependChild(p1_child1);
profile1->ToThrift(&tprofile1_v2);
averaged_profile->UpdateAverage(profile1);
deserialized_profile->Update(tprofile1_v2);
averaged_profile->GetChildren(&tmp_children);
EXPECT_EQ(2, tmp_children.size());
EXPECT_EQ("Child1", tmp_children[0]->name());
EXPECT_EQ("Child2", tmp_children[1]->name());
deserialized_profile->GetChildren(&tmp_children);
EXPECT_EQ(2, tmp_children.size());
EXPECT_EQ("Child1", tmp_children[0]->name());
EXPECT_EQ("Child2", tmp_children[1]->name());
// Test that changes in order of children is handled gracefully by preserving the
// order from the previous update. Sorting puts the children in descending total time
// order.
p1_child1->total_time_counter()->Set(1);
p1_child2->total_time_counter()->Set(2);
profile1->SortChildrenByTotalTime();
profile1->GetChildren(&tmp_children);
EXPECT_EQ("Child2", tmp_children[0]->name());
EXPECT_EQ("Child1", tmp_children[1]->name());
profile1->ToThrift(&tprofile1_v3);
averaged_profile->UpdateAverage(profile1);
deserialized_profile->Update(tprofile1_v2);
// The previous order of children that were already present is preserved.
averaged_profile->GetChildren(&tmp_children);
EXPECT_EQ(2, tmp_children.size());
EXPECT_EQ("Child1", tmp_children[0]->name());
EXPECT_EQ("Child2", tmp_children[1]->name());
deserialized_profile->GetChildren(&tmp_children);
EXPECT_EQ(2, tmp_children.size());
EXPECT_EQ("Child1", tmp_children[0]->name());
EXPECT_EQ("Child2", tmp_children[1]->name());
// Make sure we can print the profiles.
stringstream dummy;
averaged_profile->PrettyPrint(&dummy);
deserialized_profile->PrettyPrint(&dummy);
}
TEST(CountersTest, TotalTimeCounters) {
ObjectPool pool;
// Set up a three layer profile: parent -> child1 -> child2
RuntimeProfile* parent = RuntimeProfile::Create(&pool, "Parent");
RuntimeProfile* child1 = RuntimeProfile::Create(&pool, "Child1");
RuntimeProfile* child2 = RuntimeProfile::Create(&pool, "Child2");
child1->AddChild(child2);
parent->AddChild(child1);
// Part 1: Test accumulation of time up from child2 to child1 to parent
// One millisecond passes in child2
int64_t one_milli_ns = 1 * NANOS_PER_MICRO * MICROS_PER_MILLI;
child2->total_time_counter()->Add(1 * NANOS_PER_MICRO * MICROS_PER_MILLI);
parent->ComputeTimeInProfile();
EXPECT_EQ(child2->total_time(), one_milli_ns);
EXPECT_EQ(child2->local_time(), one_milli_ns);
// Child1 is a parent of child2, so it is expected to contain at least as much time
// as in child2. In this case, it is equal. However, none of the time is local.
EXPECT_EQ(child1->total_time(), one_milli_ns);
EXPECT_EQ(child1->local_time(), 0);
// The parent is in the same situation as child1
EXPECT_EQ(parent->total_time(), one_milli_ns);
EXPECT_EQ(parent->local_time(), 0);
// Time now accumulates up to child1
child1->total_time_counter()->Add(child2->total_time());
parent->ComputeTimeInProfile();
// This doesn't change anything for anyone
EXPECT_EQ(child2->total_time(), one_milli_ns);
EXPECT_EQ(child2->local_time(), one_milli_ns);
EXPECT_EQ(child1->total_time(), one_milli_ns);
EXPECT_EQ(child1->local_time(), 0);
EXPECT_EQ(parent->total_time(), one_milli_ns);
EXPECT_EQ(parent->local_time(), 0);
// Time now accumulates up to parent
parent->total_time_counter()->Add(child1->total_time());
parent->ComputeTimeInProfile();
// This doesn't change anything for the parent
EXPECT_EQ(child2->total_time(), one_milli_ns);
EXPECT_EQ(child2->local_time(), one_milli_ns);
EXPECT_EQ(child1->total_time(), one_milli_ns);
EXPECT_EQ(child1->local_time(), 0);
EXPECT_EQ(parent->total_time(), one_milli_ns);
EXPECT_EQ(parent->local_time(), 0);
// Part 2: Time accumulated in middle child
// Add 1ms to the middle child
child1->total_time_counter()->Add(one_milli_ns);
parent->ComputeTimeInProfile();
// Child2 did not change
EXPECT_EQ(child2->total_time(), one_milli_ns);
EXPECT_EQ(child2->local_time(), one_milli_ns);
// Child1 has 1ms more of total time and local time
EXPECT_EQ(child1->total_time(), 2 * one_milli_ns);
EXPECT_EQ(child1->local_time(), one_milli_ns);
// Parent has more total time, but no local time
EXPECT_EQ(parent->total_time(), 2 * one_milli_ns);
EXPECT_EQ(parent->local_time(), 0);
// Accumulate the middle child up to the parent
parent->total_time_counter()->Add(one_milli_ns);
parent->ComputeTimeInProfile();
// Doesn't change anything
EXPECT_EQ(child2->total_time(), one_milli_ns);
EXPECT_EQ(child2->local_time(), one_milli_ns);
EXPECT_EQ(child1->total_time(), 2 * one_milli_ns);
EXPECT_EQ(child1->local_time(), one_milli_ns);
EXPECT_EQ(parent->total_time(), 2 * one_milli_ns);
EXPECT_EQ(parent->local_time(), 0);
// Part 3: Time accumulated at parent
// Add 1ms to the parent
parent->total_time_counter()->Add(one_milli_ns);
parent->ComputeTimeInProfile();
// Child1 and child2 don't change
EXPECT_EQ(child2->total_time(), one_milli_ns);
EXPECT_EQ(child2->local_time(), one_milli_ns);
EXPECT_EQ(child1->total_time(), 2 * one_milli_ns);
EXPECT_EQ(child1->local_time(), one_milli_ns);
// Parent has 1ms more total time and local time
EXPECT_EQ(parent->total_time(), 3 * one_milli_ns);
EXPECT_EQ(parent->local_time(), one_milli_ns);
}
TEST(CountersTest, HighWaterMarkCounters) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::HighWaterMarkCounter* bytes_counter =
profile->AddHighWaterMarkCounter("bytes", TUnit::BYTES);
bytes_counter->Set(10);
EXPECT_EQ(bytes_counter->current_value(), 10);
EXPECT_EQ(bytes_counter->value(), 10);
bytes_counter->Add(5);
EXPECT_EQ(bytes_counter->current_value(), 15);
EXPECT_EQ(bytes_counter->value(), 15);
bytes_counter->Set(5);
EXPECT_EQ(bytes_counter->current_value(), 5);
EXPECT_EQ(bytes_counter->value(), 15);
bytes_counter->Add(3);
EXPECT_EQ(bytes_counter->current_value(), 8);
EXPECT_EQ(bytes_counter->value(), 15);
bool success = bytes_counter->TryAdd(20, 30);
EXPECT_TRUE(success);
EXPECT_EQ(bytes_counter->current_value(), 28);
EXPECT_EQ(bytes_counter->value(), 28);
success = bytes_counter->TryAdd(5, 30);
EXPECT_FALSE(success);
EXPECT_EQ(bytes_counter->current_value(), 28);
EXPECT_EQ(bytes_counter->value(), 28);
}
TEST(CountersTest, SummaryStatsCounters) {
ObjectPool pool;
RuntimeProfile* profile1 = RuntimeProfile::Create(&pool, "Profile 1");
RuntimeProfile::SummaryStatsCounter* summary_stats_counter_1 =
profile1->AddSummaryStatsCounter("summary_stats", TUnit::UNIT);
EXPECT_EQ(summary_stats_counter_1->value(), 0);
EXPECT_EQ(summary_stats_counter_1->MinValue(), numeric_limits<int64_t>::max());
EXPECT_EQ(summary_stats_counter_1->MaxValue(), numeric_limits<int64_t>::min());
summary_stats_counter_1->UpdateCounter(10);
EXPECT_EQ(summary_stats_counter_1->value(), 10);
EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
EXPECT_EQ(summary_stats_counter_1->MaxValue(), 10);
// Check that the average stays the same when updating with the same number.
summary_stats_counter_1->UpdateCounter(10);
EXPECT_EQ(summary_stats_counter_1->value(), 10);
EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
EXPECT_EQ(summary_stats_counter_1->MaxValue(), 10);
summary_stats_counter_1->UpdateCounter(40);
EXPECT_EQ(summary_stats_counter_1->value(), 20);
EXPECT_EQ(summary_stats_counter_1->MinValue(), 10);
EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);
// Verify an update with 0. This should still change the average as the number of
// samples increase
summary_stats_counter_1->UpdateCounter(0);
EXPECT_EQ(summary_stats_counter_1->value(), 15);
EXPECT_EQ(summary_stats_counter_1->MinValue(), 0);
EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);
// Verify a negative update..
summary_stats_counter_1->UpdateCounter(-40);
EXPECT_EQ(summary_stats_counter_1->value(), 4);
EXPECT_EQ(summary_stats_counter_1->MinValue(), -40);
EXPECT_EQ(summary_stats_counter_1->MaxValue(), 40);
RuntimeProfile* profile2 = RuntimeProfile::Create(&pool, "Profile 2");
RuntimeProfile::SummaryStatsCounter* summary_stats_counter_2 =
profile2->AddSummaryStatsCounter("summary_stats", TUnit::UNIT);
summary_stats_counter_2->UpdateCounter(100);
EXPECT_EQ(summary_stats_counter_2->value(), 100);
EXPECT_EQ(summary_stats_counter_2->MinValue(), 100);
EXPECT_EQ(summary_stats_counter_2->MaxValue(), 100);
TRuntimeProfileTree tprofile1;
profile1->ToThrift(&tprofile1);
// Merge profile1 and profile2 and check that profile2 is overwritten.
profile2->Update(tprofile1);
EXPECT_EQ(summary_stats_counter_2->value(), 4);
EXPECT_EQ(summary_stats_counter_2->MinValue(), -40);
EXPECT_EQ(summary_stats_counter_2->MaxValue(), 40);
}
TEST(CountersTest, DerivedCounters) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::Counter* bytes_counter =
profile->AddCounter("bytes", TUnit::BYTES);
RuntimeProfile::Counter* ticks_counter =
profile->AddCounter("ticks", TUnit::TIME_NS);
// set to 1 sec
ticks_counter->Set(1000L * 1000L * 1000L);
RuntimeProfile::DerivedCounter* throughput_counter =
profile->AddDerivedCounter("throughput", TUnit::BYTES,
bind<int64_t>(&RuntimeProfile::UnitsPerSecond, bytes_counter, ticks_counter));
bytes_counter->Set(10);
EXPECT_EQ(throughput_counter->value(), 10);
bytes_counter->Set(20);
EXPECT_EQ(throughput_counter->value(), 20);
ticks_counter->Set(ticks_counter->value() / 2);
EXPECT_EQ(throughput_counter->value(), 40);
}
TEST(CountersTest, AverageSetCounters) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::Counter* bytes_1_counter =
profile->AddCounter("bytes 1", TUnit::BYTES);
RuntimeProfile::Counter* bytes_2_counter =
profile->AddCounter("bytes 2", TUnit::BYTES);
bytes_1_counter->Set(10);
RuntimeProfile::AveragedCounter bytes_avg(TUnit::BYTES);
bytes_avg.UpdateCounter(bytes_1_counter);
// Avg of 10L
EXPECT_EQ(bytes_avg.value(), 10);
bytes_1_counter->Set(20L);
bytes_avg.UpdateCounter(bytes_1_counter);
// Avg of 20L
EXPECT_EQ(bytes_avg.value(), 20);
bytes_2_counter->Set(40L);
bytes_avg.UpdateCounter(bytes_2_counter);
// Avg of 20L and 40L
EXPECT_EQ(bytes_avg.value(), 30);
bytes_2_counter->Set(30L);
bytes_avg.UpdateCounter(bytes_2_counter);
// Avg of 20L and 30L
EXPECT_EQ(bytes_avg.value(), 25);
RuntimeProfile::Counter* double_1_counter =
profile->AddCounter("double 1", TUnit::DOUBLE_VALUE);
RuntimeProfile::Counter* double_2_counter =
profile->AddCounter("double 2", TUnit::DOUBLE_VALUE);
double_1_counter->Set(1.0f);
RuntimeProfile::AveragedCounter double_avg(TUnit::DOUBLE_VALUE);
double_avg.UpdateCounter(double_1_counter);
// Avg of 1.0f
EXPECT_EQ(double_avg.double_value(), 1.0f);
double_1_counter->Set(2.0f);
double_avg.UpdateCounter(double_1_counter);
// Avg of 2.0f
EXPECT_EQ(double_avg.double_value(), 2.0f);
double_2_counter->Set(4.0f);
double_avg.UpdateCounter(double_2_counter);
// Avg of 2.0f and 4.0f
EXPECT_EQ(double_avg.double_value(), 3.0f);
double_2_counter->Set(3.0f);
double_avg.UpdateCounter(double_2_counter);
// Avg of 2.0f and 3.0f
EXPECT_EQ(double_avg.double_value(), 2.5f);
}
TEST(CountersTest, InfoStringTest) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
EXPECT_TRUE(profile->GetInfoString("Key") == NULL);
profile->AddInfoString("Key", "Value");
const string* value = profile->GetInfoString("Key");
EXPECT_TRUE(value != NULL);
EXPECT_EQ(*value, "Value");
// Convert it to thrift
TRuntimeProfileTree tprofile;
profile->ToThrift(&tprofile);
// Convert it back
RuntimeProfile* from_thrift = RuntimeProfile::CreateFromThrift(
&pool, tprofile);
value = from_thrift->GetInfoString("Key");
EXPECT_TRUE(value != NULL);
EXPECT_EQ(*value, "Value");
// Test update.
RuntimeProfile* update_dst_profile = RuntimeProfile::Create(&pool, "Profile2");
update_dst_profile->Update(tprofile);
value = update_dst_profile->GetInfoString("Key");
EXPECT_TRUE(value != NULL);
EXPECT_EQ(*value, "Value");
// Update the original profile, convert it to thrift and update from the dst
// profile
profile->AddInfoString("Key", "NewValue");
profile->AddInfoString("Foo", "Bar");
EXPECT_EQ(*profile->GetInfoString("Key"), "NewValue");
EXPECT_EQ(*profile->GetInfoString("Foo"), "Bar");
profile->ToThrift(&tprofile);
update_dst_profile->Update(tprofile);
EXPECT_EQ(*update_dst_profile->GetInfoString("Key"), "NewValue");
EXPECT_EQ(*update_dst_profile->GetInfoString("Foo"), "Bar");
}
TEST(CountersTest, RateCounters) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::Counter* bytes_counter =
profile->AddCounter("bytes", TUnit::BYTES);
RuntimeProfile::Counter* rate_counter =
profile->AddRateCounter("RateCounter", bytes_counter);
EXPECT_TRUE(rate_counter->unit() == TUnit::BYTES_PER_SECOND);
EXPECT_EQ(rate_counter->value(), 0);
// set to 100MB. Use bigger units to avoid truncating to 0 after divides.
bytes_counter->Set(100L * 1024L * 1024L);
// Wait one second.
sleep(1);
int64_t rate = rate_counter->value();
// Stop the counter so it no longer gets updates
profile->StopPeriodicCounters();
// The rate counter is not perfectly accurate. Currently updated at 500ms intervals,
// we should have seen somewhere between 1 and 3 updates (33 - 200 MB/s)
EXPECT_GT(rate, 66 * 1024 * 1024);
EXPECT_LE(rate, 200 * 1024 * 1024);
// Wait another second. The counter has been removed. So the value should not be
// changed (much).
sleep(2);
rate = rate_counter->value();
EXPECT_GT(rate, 66 * 1024 * 1024);
EXPECT_LE(rate, 200 * 1024 * 1024);
}
TEST(CountersTest, BucketCounters) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::Counter* unit_counter =
profile->AddCounter("unit", TUnit::UNIT);
// Set the unit to 1 before sampling
unit_counter->Set(1);
// Create the bucket counters and start sampling
vector<RuntimeProfile::Counter*>* buckets =
profile->AddBucketingCounters(unit_counter, 2);
// Wait two seconds.
sleep(2);
// Stop sampling
profile->StopPeriodicCounters();
// TODO: change the value to double
// The value of buckets[0] should be zero and buckets[1] should be 1.
double val0 = (*buckets)[0]->double_value();
double val1 = (*buckets)[1]->double_value();
EXPECT_EQ(0, val0);
EXPECT_EQ(100, val1);
// Wait another second. The counter has been removed. So the value should not be
// changed (much).
sleep(2);
EXPECT_EQ(val0, (*buckets)[0]->double_value());
EXPECT_EQ(val1, (*buckets)[1]->double_value());
}
TEST(CountersTest, EventSequences) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::EventSequence* seq = profile->AddEventSequence("event sequence");
seq->MarkEvent("aaaa");
seq->MarkEvent("bbbb");
seq->MarkEvent("cccc");
vector<RuntimeProfile::EventSequence::Event> events;
seq->GetEvents(&events);
EXPECT_EQ(3, events.size());
uint64_t last_timestamp = 0;
string last_string = "";
for (const RuntimeProfile::EventSequence::Event& ev: events) {
EXPECT_TRUE(ev.second >= last_timestamp);
last_timestamp = ev.second;
EXPECT_TRUE(ev.first > last_string);
last_string = ev.first;
}
TRuntimeProfileTree thrift_profile;
profile->ToThrift(&thrift_profile);
EXPECT_TRUE(thrift_profile.nodes[0].__isset.event_sequences);
EXPECT_EQ(1, thrift_profile.nodes[0].event_sequences.size());
RuntimeProfile* reconstructed_profile =
RuntimeProfile::CreateFromThrift(&pool, thrift_profile);
last_timestamp = 0;
last_string = "";
EXPECT_EQ(NULL, reconstructed_profile->GetEventSequence("doesn't exist"));
seq = reconstructed_profile->GetEventSequence("event sequence");
EXPECT_TRUE(seq != NULL);
seq->GetEvents(&events);
EXPECT_EQ(3, events.size());
for (const RuntimeProfile::EventSequence::Event& ev: events) {
EXPECT_TRUE(ev.second >= last_timestamp);
last_timestamp = ev.second;
EXPECT_TRUE(ev.first > last_string);
last_string = ev.first;
}
}
TEST(CountersTest, UpdateEmptyEventSequence) {
// IMPALA-6824: This test makes sure that adding events to an empty event sequence does
// not crash.
ObjectPool pool;
// Create the profile to send in the update and add some events.
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::EventSequence* seq = profile->AddEventSequence("event sequence");
seq->Start();
// Sleep for 10ms to make sure the events are logged at a time > 0.
SleepForMs(10);
seq->MarkEvent("aaaa");
seq->MarkEvent("bbbb");
vector<RuntimeProfile::EventSequence::Event> events;
seq->GetEvents(&events);
EXPECT_EQ(2, events.size());
TRuntimeProfileTree thrift_profile;
profile->ToThrift(&thrift_profile);
// Create the profile that will be updated and add the empty event sequence to it.
RuntimeProfile* updated_profile = RuntimeProfile::Create(&pool, "Updated Profile");
seq = updated_profile->AddEventSequence("event sequence");
updated_profile->Update(thrift_profile);
// Verify that the events have been updated successfully.
events.clear();
seq->GetEvents(&events);
EXPECT_EQ(2, events.size());
}
void ValidateSampler(const StreamingSampler<int, 10>& sampler, int expected_num,
int expected_period, int expected_delta) {
const int* samples = NULL;
int num_samples;
int period;
samples = sampler.GetSamples(&num_samples, &period);
EXPECT_TRUE(samples != NULL);
EXPECT_EQ(num_samples, expected_num);
EXPECT_EQ(period, expected_period);
for (int i = 0; i < expected_num - 1; ++i) {
EXPECT_EQ(samples[i] + expected_delta, samples[i + 1]) << i;
}
}
TEST(CountersTest, StreamingSampler) {
StreamingSampler<int, 10> sampler;
int idx = 0;
for (int i = 0; i < 3; ++i) {
sampler.AddSample(idx++, 500);
}
ValidateSampler(sampler, 3, 500, 1);
for (int i = 0; i < 3; ++i) {
sampler.AddSample(idx++, 500);
}
ValidateSampler(sampler, 6, 500, 1);
for (int i = 0; i < 3; ++i) {
sampler.AddSample(idx++, 500);
}
ValidateSampler(sampler, 9, 500, 1);
// Added enough to cause a collapse
for (int i = 0; i < 3; ++i) {
sampler.AddSample(idx++, 500);
}
// Added enough to cause a collapse
ValidateSampler(sampler, 6, 1000, 2);
for (int i = 0; i < 3; ++i) {
sampler.AddSample(idx++, 500);
}
ValidateSampler(sampler, 7, 1000, 2);
}
// Test class to test ConcurrentStopWatch and RuntimeProfile::ConcurrentTimerCounter
// don't double count in multithread environment.
class TimerCounterTest {
public:
TimerCounterTest()
: timercounter_(TUnit::TIME_NS) {}
struct DummyWorker {
thread* thread_handle;
bool done;
DummyWorker()
: thread_handle(NULL), done(false) {}
~DummyWorker() {
Stop();
}
void Stop() {
if (!done && thread_handle != NULL) {
done = true;
thread_handle->join();
delete thread_handle;
thread_handle = NULL;
}
}
};
void Run(DummyWorker* worker) {
SCOPED_CONCURRENT_STOP_WATCH(&csw_);
SCOPED_CONCURRENT_COUNTER(&timercounter_);
while (!worker->done) {
SleepForMs(10);
// Each test case should be no more than one second.
// Consider test failed if timer is more than 3 seconds.
if (csw_.TotalRunningTime() > 6000000000) {
FAIL();
}
}
}
// Start certain number of worker threads. If interval is set, it will add some delay
// between creating worker thread.
void StartWorkers(int num, int interval) {
workers_.reserve(num);
for (int i = 0; i < num; ++i) {
workers_.push_back(DummyWorker());
DummyWorker& worker = workers_.back();
worker.thread_handle = new thread(&TimerCounterTest::Run, this, &worker);
SleepForMs(interval);
}
}
// Stop specified thread by index. if index is -1, stop all threads
void StopWorkers(int thread_index = -1) {
if (thread_index >= 0) {
workers_[thread_index].Stop();
} else {
for (int i = 0; i < workers_.size(); ++i) {
workers_[i].Stop();
}
}
}
void Reset() {
workers_.clear();
}
// Allow some timer inaccuracy (30ms) since thread join could take some time.
static const int MAX_TIMER_ERROR_NS = 30000000;
vector<DummyWorker> workers_;
ConcurrentStopWatch csw_;
RuntimeProfile::ConcurrentTimerCounter timercounter_;
};
void ValidateTimerValue(const TimerCounterTest& timer, int64_t start) {
int64_t expected_value = MonotonicStopWatch::Now() - start;
int64_t stopwatch_value = timer.csw_.TotalRunningTime();
EXPECT_GE(stopwatch_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
EXPECT_LE(stopwatch_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);
int64_t timer_value = timer.timercounter_.value();
EXPECT_GE(timer_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
EXPECT_LE(timer_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);
}
void ValidateLapTime(TimerCounterTest* timer, int64_t expected_value) {
int64_t stopwatch_value = timer->csw_.LapTime();
EXPECT_GE(stopwatch_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
EXPECT_LE(stopwatch_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);
int64_t timer_value = timer->timercounter_.LapTime();
EXPECT_GE(timer_value, expected_value - TimerCounterTest::MAX_TIMER_ERROR_NS);
EXPECT_LE(timer_value, expected_value + TimerCounterTest::MAX_TIMER_ERROR_NS);
}
TEST(TimerCounterTest, CountersTestOneThread) {
TimerCounterTest tester;
int64_t start = MonotonicStopWatch::Now();
tester.StartWorkers(1, 0);
SleepForMs(500);
ValidateTimerValue(tester, start);
tester.StopWorkers(-1);
ValidateTimerValue(tester, start);
}
TEST(TimerCounterTest, CountersTestTwoThreads) {
TimerCounterTest tester;
int64_t start = MonotonicStopWatch::Now();
tester.StartWorkers(2, 10);
SleepForMs(500);
ValidateTimerValue(tester, start);
tester.StopWorkers(-1);
ValidateTimerValue(tester, start);
}
TEST(TimerCounterTest, CountersTestRandom) {
TimerCounterTest tester;
int64_t start = MonotonicStopWatch::Now();
ValidateTimerValue(tester, start);
// First working period
tester.StartWorkers(5, 10);
ValidateTimerValue(tester, start);
SleepForMs(400);
tester.StopWorkers(2);
ValidateTimerValue(tester, start);
SleepForMs(100);
tester.StopWorkers(4);
ValidateTimerValue(tester, start);
SleepForMs(600);
tester.StopWorkers(-1);
ValidateTimerValue(tester, start);
tester.Reset();
ValidateLapTime(&tester, MonotonicStopWatch::Now() - start);
int64_t first_run_end = MonotonicStopWatch::Now();
// Adding some idle time. concurrent stopwatch and timer should not count the idle time.
SleepForMs(200);
start += MonotonicStopWatch::Now() - first_run_end;
// Second working period
tester.StartWorkers(2, 0);
// We just get lap time after first run finish. so at start of second run, expect lap time == 0
ValidateLapTime(&tester, 0);
int64_t lap_time_start = MonotonicStopWatch::Now();
SleepForMs(200);
ValidateTimerValue(tester, start);
SleepForMs(200);
tester.StopWorkers(-1);
ValidateTimerValue(tester, start);
ValidateLapTime(&tester, MonotonicStopWatch::Now() - lap_time_start);
}
TEST(TimeSeriesCounterTest, TestAddClearRace) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
int i = 0;
// Return and increment i
auto f = [&i]() { return i++; };
RuntimeProfile::TimeSeriesCounter* counter =
profile->AddChunkedTimeSeriesCounter("Counter", TUnit::UNIT, f);
// Sleep 1 second for some values to accumulate.
sleep(1);
int num_samples, period;
counter->GetSamplesTest(&num_samples, &period);
EXPECT_GT(num_samples, 0);
// Wait for more values to show up
sleep(1);
// Stop the counters. The rest of the test assumes that no new values will be added.
profile->StopPeriodicCounters();
// Clear the counter
profile->ClearChunkedTimeSeriesCounters();
// Check that clearing multiple times doesn't affect valued that have not been
// retrieved.
profile->ClearChunkedTimeSeriesCounters();
// Make sure that it still has values in it.
counter->GetSamplesTest(&num_samples, &period);
EXPECT_GT(num_samples, 0);
// Clear it again
profile->ClearChunkedTimeSeriesCounters();
// Make sure the values are gone.
counter->GetSamplesTest(&num_samples, &period);
EXPECT_EQ(num_samples, 0);
}
/// Stops the periodic counter updater in 'profile' and then clears the samples in
/// 'counter'.
void StopAndClearCounter(RuntimeProfile* profile,
RuntimeProfile::TimeSeriesCounter* counter) {
// There's a race between adding the counter and calling StopPeriodicCounters so we
// sleep here to make sure we exercise the code that handles the race.
sleep(1);
profile->StopPeriodicCounters();
// Reset the counter state by reading and clearing its samples.
int num_samples = 0;
int result_period_unused = 0;
counter->GetSamplesTest(&num_samples, &result_period_unused);
ASSERT_GT(num_samples, 0);
profile->ClearChunkedTimeSeriesCounters();
// Ensure clean state.
counter->GetSamplesTest(&num_samples, &result_period_unused);
ASSERT_EQ(num_samples, 0);
}
/// Tests that ChunkedTimeSeriesCounters are bounded by a maximum size.
TEST(TimeSeriesCounterTest, TestMaximumSize) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
const int test_period = FLAGS_periodic_counter_update_period_ms;
// Add a counter with a sample function that counts up, starting from 0.
int value = 0;
auto sample_fn = [&value]() { return value++; };
RuntimeProfile::TimeSeriesCounter* counter =
profile->AddChunkedTimeSeriesCounter("TestCounter", TUnit::UNIT, sample_fn);
// Stop counter updates from interfering with the rest of the test.
StopAndClearCounter(profile, counter);
// Reset value after previous values have been retrieved.
value = 0;
int64_t max_size = 10 * FLAGS_status_report_interval_ms / test_period;
for (int i = 0; i < 10 + max_size; ++i) counter->AddSample(test_period);
int num_samples = 0;
int result_period = 0;
// Retrieve and validate samples.
const int64_t* samples = counter->GetSamplesTest(&num_samples, &result_period);
ASSERT_EQ(num_samples, max_size);
// No resampling happens with ChunkedTimeSeriesCounter.
ASSERT_EQ(result_period, test_period);
// First 10 samples have been truncated
ASSERT_EQ(samples[0], 10);
}
/// Test parameter class that helps to test time series resampling during profile pretty
/// printing with a varying number of test samples.
struct TimeSeriesTestParam {
TimeSeriesTestParam(int num_samples, vector<const char*> expected)
: num_samples(num_samples), expected(expected) {}
int num_samples;
vector<const char*> expected;
// Used by gtest to print values of this struct
friend std::ostream& operator<<(std::ostream& os, const TimeSeriesTestParam& p) {
return os << "num_samples: " << p.num_samples << endl;
}
};
class TimeSeriesCounterResampleTest : public testing::TestWithParam<TimeSeriesTestParam> {
};
/// Tests that pretty-printing a ChunkedTimeSeriesCounter limits the number or printed
/// samples to 64 or lower.
TEST_P(TimeSeriesCounterResampleTest, TestPrettyPrint) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
const TimeSeriesTestParam& param = GetParam();
const int test_period = FLAGS_periodic_counter_update_period_ms;
// Add a counter with a sample function that counts up, starting from 0.
int value = 0;
auto sample_fn = [&value]() { return value++; };
// We increase the value of this flag to allow the counter to store enough samples.
FLAGS_status_report_interval_ms = 50000;
RuntimeProfile::TimeSeriesCounter* counter =
profile->AddChunkedTimeSeriesCounter("TestCounter", TUnit::UNIT, sample_fn);
// Stop counter updates from interfering with the rest of the test.
StopAndClearCounter(profile, counter);
// Reset value after previous values have been retrieved.
value = 0;
for (int i = 0; i < param.num_samples; ++i) counter->AddSample(test_period);
int num_samples = 0;
int result_period = 0;
// Retrieve and validate samples.
const int64_t* samples = counter->GetSamplesTest(&num_samples, &result_period);
ASSERT_EQ(num_samples, param.num_samples);
// No resampling happens with ChunkedTimeSeriesCounter.
ASSERT_EQ(result_period, test_period);
for (int i = 0; i < param.num_samples; ++i) ASSERT_EQ(samples[i], i);
stringstream pretty;
profile->PrettyPrint(&pretty);
const string pretty_str = pretty.str();
for (const char* e : param.expected) EXPECT_STR_CONTAINS(pretty_str, e);
}
INSTANTIATE_TEST_CASE_P(VariousNumbers, TimeSeriesCounterResampleTest,
::testing::Values(
TimeSeriesTestParam(64, {"TestCounter (500.000ms): 0, 1, 2, 3", "61, 62, 63"}),
TimeSeriesTestParam(65, {"TestCounter (1s000ms): 0, 2, 4, 6,",
"60, 62, 64 (Showing 33 of 65 values from Thrift Profile)"}),
TimeSeriesTestParam(80, {"TestCounter (1s000ms): 0, 2, 4, 6,",
"74, 76, 78 (Showing 40 of 80 values from Thrift Profile)"}),
TimeSeriesTestParam(127, {"TestCounter (1s000ms): 0, 2, 4, 6,",
"122, 124, 126 (Showing 64 of 127 values from Thrift Profile)"}),
TimeSeriesTestParam(128, {"TestCounter (1s000ms): 0, 2, 4, 6,",
"122, 124, 126 (Showing 64 of 128 values from Thrift Profile)"}),
TimeSeriesTestParam(129, {"TestCounter (1s500ms): 0, 3, 6, 9,",
"120, 123, 126 (Showing 43 of 129 values from Thrift Profile)"})
));
// Tests that the __isset field for TRuntimeProfileNode.node_metadata is set correctly
// (IMPALA-8252).
TEST(ToThrift, NodeMetadataIsSetCorrectly) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
TRuntimeProfileTree thrift_profile;
profile->ToThrift(&thrift_profile);
// Profile is empty, expect 0 nodes
EXPECT_EQ(thrift_profile.nodes.size(), 1);
EXPECT_FALSE(thrift_profile.nodes[0].__isset.node_metadata);
// Set the plan node ID and make sure that the field is marked correctly
profile->SetPlanNodeId(1);
profile->ToThrift(&thrift_profile);
EXPECT_TRUE(thrift_profile.nodes[0].__isset.node_metadata);
}
TEST(ToJson, RuntimeProfileToJsonTest) {
ObjectPool pool;
RuntimeProfile* profile_a = RuntimeProfile::Create(&pool, "ProfileA");
RuntimeProfile* profile_a1 = RuntimeProfile::Create(&pool, "ProfileA1");
RuntimeProfile* profile_ab = RuntimeProfile::Create(&pool, "ProfileAb");
RuntimeProfile::Counter* counter_a;
// Initialize for further validation
profile_a->AddChild(profile_a1);
profile_a->AddChild(profile_ab);
profile_a->AddInfoString("Key", "Value");
counter_a = profile_a->AddCounter("A", TUnit::UNIT);
counter_a->Set(1);
RuntimeProfile::HighWaterMarkCounter* high_water_counter =
profile_a->AddHighWaterMarkCounter("high_water_counter", TUnit::BYTES);
high_water_counter->Set(10);
high_water_counter->Add(10);
high_water_counter->Set(10);
RuntimeProfile::SummaryStatsCounter* summary_stats_counter =
profile_a->AddSummaryStatsCounter("summary_stats_counter", TUnit::TIME_NS);
summary_stats_counter->UpdateCounter(10);
summary_stats_counter->UpdateCounter(20);
// Serialize to json
rapidjson::Document doc(rapidjson::kObjectType);
profile_a->ToJson(&doc);
rapidjson::Value& content = doc["contents"];
// Check profile correct
EXPECT_EQ("ProfileA", content["profile_name"]);
EXPECT_EQ("ProfileA1", content["child_profiles"][0]["profile_name"]);
EXPECT_EQ("ProfileAb", content["child_profiles"][1]["profile_name"]);
// Check Info String correct
EXPECT_EQ(1, content["info_strings"].Size());
EXPECT_EQ("Key", content["info_strings"][0]["key"]);
EXPECT_EQ("Value", content["info_strings"][0]["value"]);
// Check counter value matches
EXPECT_EQ(2, content["counters"].Size());
for (auto& itr : content["counters"].GetArray()) {
// check normal Counter
if (itr["counter_name"] == "A") {
EXPECT_EQ(1, itr["value"].GetInt());
EXPECT_EQ("UNIT", itr["unit"]);
EXPECT_EQ("Counter", itr["kind"]);
}// check HighWaterMarkCounter
else if (itr["counter_name"] == "high_water_counter") {
EXPECT_EQ(20, itr["value"].GetInt());
EXPECT_EQ("BYTES", itr["unit"]);
EXPECT_EQ("HighWaterMarkCounter", itr["kind"]);
} else {
DCHECK(false);
}
}
// Check SummaryStatsCounter
EXPECT_EQ(1, content["summary_stats_counters"].Size());
for (auto& itr : content["summary_stats_counters"].GetArray()) {
if (itr["counter_name"] == "summary_stats_counter") {
EXPECT_EQ(10, itr["min"].GetInt());
EXPECT_EQ(20, itr["max"].GetInt());
EXPECT_EQ(15, itr["avg"].GetInt());
EXPECT_EQ(2, itr["num_of_samples"].GetInt());
EXPECT_EQ("TIME_NS", itr["unit"]);
EXPECT_TRUE(!itr.HasMember("kind"));
}
}
}
// Test when some fields are not set. ToJson will not add them as a member
TEST(ToJson, EmptyTest) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
// Serialize to json
rapidjson::Document doc(rapidjson::kObjectType);
profile->ToJson(&doc);
rapidjson::Value& content = doc["contents"];
EXPECT_EQ("Profile", content["profile_name"]);
EXPECT_TRUE(content.HasMember("num_children"));
// Empty profile should not have following members
EXPECT_TRUE(!content.HasMember("info_strings"));
EXPECT_TRUE(!content.HasMember("event_sequences"));
EXPECT_TRUE(!content.HasMember("counters"));
EXPECT_TRUE(!content.HasMember("summary_stats_counters"));
EXPECT_TRUE(!content.HasMember("time_series_counters"));
EXPECT_TRUE(!content.HasMember("child_profiles"));
}
TEST(ToJson, EventSequenceToJsonTest) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
RuntimeProfile::EventSequence* seq = profile->AddEventSequence("event sequence");
seq->MarkEvent("aaaa");
seq->MarkEvent("bbbb");
seq->MarkEvent("cccc");
// Serialize to json
rapidjson::Document doc(rapidjson::kObjectType);
rapidjson::Value event_sequence_json(rapidjson::kObjectType);
seq->ToJson(doc, &event_sequence_json);
EXPECT_EQ(0, event_sequence_json["offset"].GetInt());
uint64_t last_timestamp = 0;
string last_string = "";
for (auto& itr : event_sequence_json["events"].GetArray()) {
EXPECT_TRUE(itr["timestamp"].GetInt() >= last_timestamp);
last_timestamp = itr["timestamp"].GetInt();
string label = string(itr["label"].GetString());
EXPECT_TRUE(label > last_string);
last_string = label;
}
}
TEST(ToJson, TimeSeriesCounterToJsonTest) {
ObjectPool pool;
RuntimeProfile* profile = RuntimeProfile::Create(&pool, "Profile");
// 1. TimeSeriesCounter should be empty
rapidjson::Document doc(rapidjson::kObjectType);
profile->ToJson(&doc);
EXPECT_TRUE(!doc["contents"].HasMember("time_series_counters"));
// 2. Check Serialize to json
const int test_period = FLAGS_periodic_counter_update_period_ms;
// Add a counter with a sample function that counts up, starting from 0.
int value = 0;
auto sample_fn = [&value]() { return value++; };
// We increase the value of this flag to allow the counter to store enough samples.
FLAGS_status_report_interval_ms = 50000;
RuntimeProfile::TimeSeriesCounter* counter =
profile->AddChunkedTimeSeriesCounter("TimeSeriesCounter", TUnit::UNIT, sample_fn);
RuntimeProfile::TimeSeriesCounter* counter2 =
profile->AddSamplingTimeSeriesCounter("SamplingCounter", TUnit::UNIT, sample_fn);
// Stop counter updates from interfering with the rest of the test.
StopAndClearCounter(profile, counter);
// Reset value after previous values have been retrieved.
value = 0;
for (int i = 0; i < 64; ++i) counter->AddSample(test_period);
value = 0;
for (int i = 0; i < 80; ++i) counter2->AddSample(test_period);
profile->ToJson(&doc);
EXPECT_STR_CONTAINS(
doc["contents"]["time_series_counters"][1]["data"].GetString(), "0,1,2,3,4");
EXPECT_STR_CONTAINS(
doc["contents"]["time_series_counters"][1]["data"].GetString(), "60,61,62,63");
EXPECT_STR_CONTAINS(
doc["contents"]["time_series_counters"][0]["data"].GetString(), "0,2,4,6");
EXPECT_STR_CONTAINS(
doc["contents"]["time_series_counters"][0]["data"].GetString(), "72,74,76,78");
}
} // namespace impala