Skip to content
This repository has been archived by the owner on Aug 19, 2019. It is now read-only.

Handle deletion events from watch and purge deleted entries. #62

Merged
merged 4 commits into from
Feb 26, 2018
Merged

Handle deletion events from watch and purge deleted entries. #62

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
1c2a83c
Handle deletion watch events.
igorpeshansky Feb 25, 2018
aeb1c7d
Factor out the common watch event callback.
igorpeshansky Feb 25, 2018
13fa66f
Purge deleted entries after reporting metadata to the API.
igorpeshansky Feb 26, 2018
9c46711
There may not be a container id even if container status is present.
igorpeshansky Feb 26, 2018
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
57 changes: 43 additions & 14 deletions src/api_server.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ class MetadataApiServer {

class MetadataReporter {
public:
MetadataReporter(const MetadataAgent& agent, double period_s);
MetadataReporter(MetadataAgent* agent, double period_s);
~MetadataReporter();

private:
Expand All @@ -79,7 +79,7 @@ class MetadataReporter {
std::map<MonitoredResource, MetadataAgent::Metadata>&& metadata)
throw (boost::system::system_error);

const MetadataAgent& agent_;
MetadataAgent* agent_;
Environment environment_;
OAuth2 auth_;
// The reporting period in seconds.
Expand Down Expand Up @@ -151,9 +151,9 @@ MetadataApiServer::~MetadataApiServer() {
}
}

MetadataReporter::MetadataReporter(const MetadataAgent& agent, double period_s)
MetadataReporter::MetadataReporter(MetadataAgent* agent, double period_s)
: agent_(agent),
environment_(agent.config_),
environment_(agent->config_),
auth_(environment_),
period_(period_s),
reporter_thread_(&MetadataReporter::ReportMetadata, this) {}
Expand All @@ -169,14 +169,17 @@ void MetadataReporter::ReportMetadata() {
std::this_thread::sleep_for(std::chrono::seconds(3));
// TODO: Do we need to be able to stop this?
while (true) {
if (agent_.config_.VerboseLogging()) {
if (agent_->config_.VerboseLogging()) {
LOG(INFO) << "Sending metadata request to server";
}
try {
SendMetadata(agent_.GetMetadataMap());
if (agent_.config_.VerboseLogging()) {
SendMetadata(agent_->GetMetadataMap());
if (agent_->config_.VerboseLogging()) {
LOG(INFO) << "Metadata request sent successfully";
}
if (agent_->config_.MetadataReporterPurgeDeleted()) {
agent_->PurgeDeletedEntries();
}
} catch (const boost::system::system_error& e) {
LOG(ERROR) << "Metadata request unsuccessful: " << e.what();
}
Expand Down Expand Up @@ -229,20 +232,20 @@ void MetadataReporter::SendMetadata(
std::map<MonitoredResource, MetadataAgent::Metadata>&& metadata)
throw (boost::system::system_error) {
if (metadata.empty()) {
if (agent_.config_.VerboseLogging()) {
if (agent_->config_.VerboseLogging()) {
LOG(INFO) << "No data to send";
}
return;
}

if (agent_.config_.VerboseLogging()) {
if (agent_->config_.VerboseLogging()) {
LOG(INFO) << "Sending request to the server";
}
const std::string project_id = environment_.NumericProjectId();
// The endpoint template is expected to be of the form
// "https://stackdriver.googleapis.com/.../projects/{{project_id}}/...".
const std::string endpoint =
format::Substitute(agent_.config_.MetadataIngestionEndpointFormat(),
format::Substitute(agent_->config_.MetadataIngestionEndpointFormat(),
{{"project_id", project_id}});
const std::string auth_header = auth_.GetAuthHeaderValue();

Expand All @@ -252,7 +255,7 @@ void MetadataReporter::SendMetadata(
const int empty_size = empty_request->ToString().size();

const int limit_bytes =
agent_.config_.MetadataIngestionRequestSizeLimitBytes();
agent_->config_.MetadataIngestionRequestSizeLimitBytes();
int total_size = empty_size;

std::vector<json::value> entries;
Expand Down Expand Up @@ -281,7 +284,7 @@ void MetadataReporter::SendMetadata(
}
if (total_size + size > limit_bytes) {
SendMetadataRequest(std::move(entries), endpoint, auth_header,
agent_.config_.VerboseLogging());
agent_->config_.VerboseLogging());
entries.clear();
total_size = empty_size;
}
Expand All @@ -290,7 +293,7 @@ void MetadataReporter::SendMetadata(
}
if (!entries.empty()) {
SendMetadataRequest(std::move(entries), endpoint, auth_header,
agent_.config_.VerboseLogging());
agent_->config_.VerboseLogging());
}
}

Expand Down Expand Up @@ -345,12 +348,38 @@ std::map<MonitoredResource, MetadataAgent::Metadata>
return result;
}

void MetadataAgent::PurgeDeletedEntries() {
std::lock_guard<std::mutex> lock(metadata_mu_);

for (auto it = metadata_map_.begin(); it != metadata_map_.end(); ) {
const MonitoredResource& resource = it->first;
const Metadata& entry = it->second;
if (entry.is_deleted) {
if (config_.VerboseLogging()) {
LOG(INFO) << "Purging metadata entry " << resource << "->{"
<< "version: " << entry.version << ", "
<< "is_deleted: " << entry.is_deleted << ", "
<< "created_at: " << rfc3339::ToString(entry.created_at)
<< ", "
<< "collected_at: " << rfc3339::ToString(entry.collected_at)
<< ", "
<< "metadata: " << *entry.metadata << ", "
<< "ignore: " << entry.ignore
<< "}";
}
it = metadata_map_.erase(it);
} else {
++it;
}
}
}

void MetadataAgent::start() {
metadata_api_server_.reset(new MetadataApiServer(
*this, config_.MetadataApiNumThreads(), "0.0.0.0",
config_.MetadataApiPort()));
reporter_.reset(new MetadataReporter(
*this, config_.MetadataReporterIntervalSeconds()));
this, config_.MetadataReporterIntervalSeconds()));
}

}
1 change: 1 addition & 0 deletions src/api_server.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ class MetadataAgent {
friend class MetadataReporter;

std::map<MonitoredResource, Metadata> GetMetadataMap() const;
void PurgeDeletedEntries();

const MetadataAgentConfiguration& config_;

Expand Down
6 changes: 6 additions & 0 deletions src/configuration.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ constexpr const int kMetadataApiDefaultNumThreads = 3;
constexpr const int kMetadataApiDefaultPort = 8000;
constexpr const char kMetadataApiDefaultResourceTypeSeparator[] = ".";
constexpr const int kMetadataReporterDefaultIntervalSeconds = 60;
constexpr const int kMetadataReporterDefaultPurgeDeleted = false;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we be defaulting to true?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd rather put this through a few more tests. It can be overridden in the config map if necessary.

constexpr const char kMetadataIngestionDefaultEndpointFormat[] =
"https://stackdriver.googleapis.com/v1beta2/projects/{{project_id}}"
"/resourceMetadata:batchUpdate";
Expand Down Expand Up @@ -70,6 +71,8 @@ MetadataAgentConfiguration::MetadataAgentConfiguration()
kMetadataApiDefaultResourceTypeSeparator),
metadata_reporter_interval_seconds_(
kMetadataReporterDefaultIntervalSeconds),
metadata_reporter_purge_deleted_(
kMetadataReporterDefaultPurgeDeleted),
metadata_ingestion_endpoint_format_(
kMetadataIngestionDefaultEndpointFormat),
metadata_ingestion_request_size_limit_bytes_(
Expand Down Expand Up @@ -147,6 +150,9 @@ void MetadataAgentConfiguration::ParseConfigFile(const std::string& filename) {
metadata_reporter_interval_seconds_ =
config["MetadataReporterIntervalSeconds"].as<int>(
kMetadataReporterDefaultIntervalSeconds);
metadata_reporter_purge_deleted_ =
config["MetadataReporterPurgeDeleted"].as<bool>(
kMetadataReporterDefaultPurgeDeleted);
metadata_ingestion_endpoint_format_ =
config["MetadataIngestionEndpointFormat"].as<std::string>(
kMetadataIngestionDefaultEndpointFormat);
Expand Down
5 changes: 5 additions & 0 deletions src/configuration.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,10 @@ class MetadataAgentConfiguration {
std::lock_guard<std::mutex> lock(mutex_);
return metadata_reporter_interval_seconds_;
}
bool MetadataReporterPurgeDeleted() const {
std::lock_guard<std::mutex> lock(mutex_);
return metadata_reporter_purge_deleted_;
}
const std::string& MetadataIngestionEndpointFormat() const {
std::lock_guard<std::mutex> lock(mutex_);
return metadata_ingestion_endpoint_format_;
Expand Down Expand Up @@ -145,6 +149,7 @@ class MetadataAgentConfiguration {
int metadata_api_port_;
std::string metadata_api_resource_type_separator_;
int metadata_reporter_interval_seconds_;
bool metadata_reporter_purge_deleted_;
std::string metadata_ingestion_endpoint_format_;
int metadata_ingestion_request_size_limit_bytes_;
std::string metadata_ingestion_raw_content_version_;
Expand Down
Loading