archive/netdata_netdata
0
0
Fork 0
mirror of https://github.com/netdata/netdata.git synced 2025-04-28 06:32:30 +00:00
Code Issues Projects Releases Wiki Activity
netdata_netdata/web/api/queries/query.c
vkalintiris 9ed4cea590
Anomaly Detection MVP () 
* Add support for feature extraction and K-Means clustering.

This patch adds support for performing feature extraction and running the
K-Means clustering algorithm on the extracted features.

We use the open-source dlib library to compute the K-Means clustering
centers, which has been added as a new git submodule.

The build system has been updated to recognize two new options:

    1) --enable-ml: build an agent with ml functionality, and
    2) --enable-ml-tests: support running tests with the `-W mltest`
       option in netdata.

The second flag is meant only for internal use. To build tests successfully,
you need to install the GoogleTest framework on your machine.

* Boilerplate code to track hosts/dims and init ML config options.

A new opaque pointer field is added to the database's host and dimension
data structures. The fields point to C++ wrapper classes that will be used
to store ML-related information in follow-up patches.

The ML functionality needs to iterate all tracked dimensions twice per
second. To avoid locking the entire DB multiple times, we use a
separate dictionary to add/remove dimensions as they are created/deleted
by the database.

A global configuration object is initialized during the startup of the
agent. It will allow our users to specify ML-related configuration
options, eg. hosts/charts to skip from training, etc.

* Add support for training and prediction of dimensions.

Every new host spawns a training thread which is used to train the model
of each dimension.

Training of dimensions is done in a non-batching mode in order to avoid
impacting the generated ML model by the CPU, RAM and disk utilization of
the training code itself.

For performance reasons, prediction is done at the time a new value
is pushed in the database. The alternative option, ie. maintaining a
separate thread for prediction, would be ~3-4x times slower and would
increase locking contention considerably.

For similar reasons, we use a custom function to unpack storage_numbers
into doubles, instead of long doubles.

* Add data structures required by the anomaly detector.

This patch adds two data structures that will be used by the anomaly
detector in follow-up patches.

The first data structure is a circular bit buffer which is being used to
count the number of set bits over time.

The second data structure represents an expandable, rolling window that
tracks set/unset bits. It is explicitly modeled as a finite-state
machine in order to make the anomaly detector's behaviour easier to test
and reason about.

* Add anomaly detection thread.

This patch creates a new anomaly detection thread per host. Each thread
maintains a BitRateWindow which is updated every second based on the
anomaly status of the correspondent host.

Based on the updated status of the anomaly window, we can identify the
existence/absence of an anomaly event, it's start/end time and the
dimensions that participate in it.

* Create/insert/query anomaly events from Sqlite DB.

* Create anomaly event endpoints.

This patch adds two endpoints to expose information about anomaly
events. The first endpoint returns the list of anomalous events within a
specified time range. The second endpoint provides detailed information
about a single anomaly event, ie. the list of anomalous dimensions in
that event along with their anomaly rate.

The `anomaly-bit` option has been added to the `/data` endpoint in order
to allow users to get the anomaly status of individual dimensions per
second.

* Fix build failures on Ubuntu 16.04 & CentOS 7.

These distros do not have toolchains with C++11 enabled by default.
Replacing nullptr with NULL should be fix the build problems on these
platforms when the ML feature is not enabled.

* Fix `make dist` to include ML makefiles and dlib sources.

Currently, we add ml/kmeans/dlib to EXTRA_DIST. We might want to
generate an explicit list of source files in the future, in order to
bring down the generated archive's file size.

* Small changes to make the LGTM & Codacy bots happy.

- Cast unused result of function calls to void.
- Pass a const-ref string to Database's constructor.
- Reduce the scope of a local variable in the anomaly detector.

* Add user configuration option to enable/disable anomaly detection.

* Do not log dimension-specific operations.

Training and prediction operations happen every second for each
dimension. In prep for making this PR easier to run anomaly detection
for many charts & dimensions, I've removed logs that would cause log
flooding.

* Reset dimensions' bit counter when not above anomaly rate threshold.

* Update the default config options with real values.

With this patch the default configuration options will match the ones
we want our users to use by default.

* Update conditions for creating new ML dimensions.

1. Skip dimensions with update_every != 1,
2. Skip dimensions that come from the ML charts.

With this filtering in place, any configuration value for the
relevant simple_pattern expressions will work correctly.

* Teach buildinfo{,json} about the ML feature.

* Set --enable-ml by default in the configuration options.

This patch is only meant for testing the building of the ML functionality
on Github. It will be reverted once tests pass successfully.

* Minor build system fixes.

- Add path to json header
- Enable C++ linker when ML functionality is enabled
- Rename ml/ml-dummy.cc to ml/ml-dummy.c

* Revert "Set --enable-ml by default in the configuration options."

This reverts commit 28206952a59a577675c86194f2590ec63b60506c.

We pass all Github checks when building the ML functionality, except for
those that run on CentOS 7 due to not having a C++11 toolchain.

* Check for missing dlib and nlohmann files.

We simply check the single-source files upon which our build system
depends. If they are missing, an error message notifies the user
about missing git submodules which are required for the ML
functionality.

* Allow users to specify the maximum number of KMeans iterations.

* Use dlib v19.10

v19.22 broke compatibility with CentOS 7's g++. Development of the
anomaly detection used v19.10, which is the version used by most Debian and
Ubuntu distribution versions that are not past EOL.

No observable performance improvements/regressions specific to the K-Means
algorithm occur between the two versions.

* Detect and use the -std=c++11 flag when building anomaly detection.

This patch automatically adds the -std=c++11 when building netdata
with the ML functionality, if it's supported by the user's toolchain.

With this change we are able to build the agent correctly on CentOS 7.

* Restructure configuration options.

- update default values,
- clamp values to min/max defaults,
- validate and identify conflicting values.

* Add update_every configuration option.

Considerring that the MVP does not support per host configuration
options, the update_every option will be used to filter hosts to train.

With this change anomaly detection will be supported on:

    - Single nodes with update_every != 1, and
    - Children nodes with a common update_every value that might differ from
      the value of the parent node.

* Reorganize anomaly detection charts.

This follows Andrew's suggestion to have four charts to show the number
of anomalous/normal dimensions, the anomaly rate, the detector's window
length, and the events that occur in the prediction step.

Context and family values, along with the necessary information in the
dashboard_info.js file, will be updated in a follow-up commit.

* Do not dump anomaly event info in logs.

* Automatically handle low "train every secs" configuration values.

If a user specifies a very low value for the "train every secs", then
it is possible that the time it takes to train a dimension is higher
than the its allotted time.

In that case, we want the training thread to:

    - Reduce it's CPU usage per second, and
    - Allow the prediction thread to proceed.

We achieve this by limiting the training time of a single dimension to
be equal to half the time allotted to it. This means, that the training
thread will never consume more than 50% of a single core.

* Automatically detect if ML functionality should be enabled.

With these changes, we enable ML if:

    - The user has not explicitly specified --disable-ml, and
    - Git submodules have been checked out properly, and
    - The toolchain supports C++11.

If the user has explicitly specified --enable-ml, the build fails if
git submodules are missing, or the toolchain does not support C++11.

* Disable anomaly detection by default.

* Do not update charts in locked region.

* Cleanup code reading configuration options.

* Enable C++ linker when building ML.

* Disable ML functionality for CMake builds.

* Skip LGTM for dlib and nlohmann libraries.

* Do not build ML if libuuid is missing.

* Fix dlib path in LGTM's yaml config file.

* Add chart to track duration of prediction step.

* Add chart to track duration of training step.

* Limit the number dimensions in an anomaly event.

This will ensure our JSON results won't grow without any limit. The
default ML configuration options, train approximately ~1700 dimensions
in a newly-installed Netdata agent. The hard-limit is set to 2000
dimensions which:

    - Is well above the default number of dimensions we train,
    - If it is ever reached it means that the user had accidentaly a
      very low anomaly rate threshold, and
    - Considering that we sort the result by anomaly score, the cutoff
      dimensions will be the less anomalous, ie. the least important to
      investigate.

* Add information about the ML charts.

* Update family value in ML charts.

This fix will allow us to show the individual charts in the RHS Anomaly
Detection submenu.

* Rename chart type

s/anomalydetection/anomaly_detection/g

* Expose ML feat in /info endpoint.

* Export ML config through /info endpoint.

* Fix CentOS 7 build.

* Reduce the critical region of a host's lock.

Before this change, each host had a single, dedicated lock to protect
its map of dimensions from adding/deleting new dimensions while training
and detecting anomalies. This was problematic because training of a
single dimension can take several seconds in nodes that are under heavy
load.

After this change, the host's lock protects only the insertion/deletion
of new dimensions, and the prediction step. For the training of dimensions
we use a dedicated lock per dimension, which is responsible for protecting
the dimension from deletion while training.

Prediction is fast enough, even on slow machines or under heavy load,
which allows us to use the host's main lock and avoid increasing the
complexity of our implementation in the anomaly detector.

* Improve the way we are tracking anomaly detector's performance.

This change allows us to:

    - track the total training time per update_every period,
    - track the maximum training time of a single dimension per
      update_every period, and
    - export the current number of total, anomalous, normal dimensions
      to the /info endpoint.

Also, now that we use dedicated locks per dimensions, we can train under
heavy load continuously without having to sleep in order to yield the
training thread and allow the prediction thread to progress.

* Use samples instead of seconds in ML configuration.

This commit changes the way we are handling input ML configuration
options from the user. Instead of treating values as seconds, we
interpret all inputs as number of update_every periods. This allows
us to enable anomaly detection on hosts that have update_every != 1
second, and still produce a model for training/prediction & detection
that behaves in an expected way.

Tested by running anomaly detection on an agent with update_every = [1,
2, 4] seconds.

* Remove unecessary log message in detection thread

* Move ML configuration to global section.

* Update web/gui/dashboard_info.js

Co-authored-by: Andrew Maguire <andrewm4894@gmail.com>

* Fix typo

Co-authored-by: Andrew Maguire <andrewm4894@gmail.com>

* Rebase.

* Use negative logic for anomaly bit.

* Add info for prediction_stats and training_stats charts.

* Disable ML on PPC64EL.

The CI test fails with -std=c++11 and requires -std=gnu++11 instead.
However, it's not easy to quickly append the required flag to CXXFLAGS.
For the time being, simply disable ML on PPC64EL and if any users
require this functionality we can fix it in the future.

* Add comment on why we disable ML on PPC64EL.

Co-authored-by: Andrew Maguire <andrewm4894@gmail.com>
2021-10-27 09:26:21 +03:00

1659 lines
67 KiB
C
Raw Blame History

// SPDX-License-Identifier: GPL-3.0-or-later
#include "query.h"
#include "web/api/formatters/rrd2json.h"
#include "rrdr.h"
#include "average/average.h"
#include "incremental_sum/incremental_sum.h"
#include "max/max.h"
#include "median/median.h"
#include "min/min.h"
#include "sum/sum.h"
#include "stddev/stddev.h"
#include "ses/ses.h"
#include "des/des.h"
// ----------------------------------------------------------------------------
static struct {
const char *name;
uint32_t hash;
RRDR_GROUPING value;
// One time initialization for the module.
// This is called once, when netdata starts.
void (*init)(void);
// Allocate all required structures for a query.
// This is called once for each netdata query.
void *(*create)(struct rrdresult *r);
// Cleanup collected values, but don't destroy the structures.
// This is called when the query engine switches dimensions,
// as part of the same query (so same chart, switching metric).
void (*reset)(struct rrdresult *r);
// Free all resources allocated for the query.
void (*free)(struct rrdresult *r);
// Add a single value into the calculation.
// The module may decide to cache it, or use it in the fly.
void (*add)(struct rrdresult *r, calculated_number value);
// Generate a single result for the values added so far.
// More values and points may be requested later.
// It is up to the module to reset its internal structures
// when flushing it (so for a few modules it may be better to
// continue after a flush as if nothing changed, for others a
// cleanup of the internal structures may be required).
calculated_number (*flush)(struct rrdresult *r, RRDR_VALUE_FLAGS *rrdr_value_options_ptr);
} api_v1_data_groups[] = {
{.name = "average",
.hash = 0,
.value = RRDR_GROUPING_AVERAGE,
.init = NULL,
.create= grouping_create_average,
.reset = grouping_reset_average,
.free = grouping_free_average,
.add = grouping_add_average,
.flush = grouping_flush_average
},
{.name = "mean", // alias on 'average'
.hash = 0,
.value = RRDR_GROUPING_AVERAGE,
.init = NULL,
.create= grouping_create_average,
.reset = grouping_reset_average,
.free = grouping_free_average,
.add = grouping_add_average,
.flush = grouping_flush_average
},
{.name = "incremental_sum",
.hash = 0,
.value = RRDR_GROUPING_INCREMENTAL_SUM,
.init = NULL,
.create= grouping_create_incremental_sum,
.reset = grouping_reset_incremental_sum,
.free = grouping_free_incremental_sum,
.add = grouping_add_incremental_sum,
.flush = grouping_flush_incremental_sum
},
{.name = "incremental-sum",
.hash = 0,
.value = RRDR_GROUPING_INCREMENTAL_SUM,
.init = NULL,
.create= grouping_create_incremental_sum,
.reset = grouping_reset_incremental_sum,
.free = grouping_free_incremental_sum,
.add = grouping_add_incremental_sum,
.flush = grouping_flush_incremental_sum
},
{.name = "median",
.hash = 0,
.value = RRDR_GROUPING_MEDIAN,
.init = NULL,
.create= grouping_create_median,
.reset = grouping_reset_median,
.free = grouping_free_median,
.add = grouping_add_median,
.flush = grouping_flush_median
},
{.name = "min",
.hash = 0,
.value = RRDR_GROUPING_MIN,
.init = NULL,
.create= grouping_create_min,
.reset = grouping_reset_min,
.free = grouping_free_min,
.add = grouping_add_min,
.flush = grouping_flush_min
},
{.name = "max",
.hash = 0,
.value = RRDR_GROUPING_MAX,
.init = NULL,
.create= grouping_create_max,
.reset = grouping_reset_max,
.free = grouping_free_max,
.add = grouping_add_max,
.flush = grouping_flush_max
},
{.name = "sum",
.hash = 0,
.value = RRDR_GROUPING_SUM,
.init = NULL,
.create= grouping_create_sum,
.reset = grouping_reset_sum,
.free = grouping_free_sum,
.add = grouping_add_sum,
.flush = grouping_flush_sum
},
// standard deviation
{.name = "stddev",
.hash = 0,
.value = RRDR_GROUPING_STDDEV,
.init = NULL,
.create= grouping_create_stddev,
.reset = grouping_reset_stddev,
.free = grouping_free_stddev,
.add = grouping_add_stddev,
.flush = grouping_flush_stddev
},
{.name = "cv", // coefficient variation is calculated by stddev
.hash = 0,
.value = RRDR_GROUPING_CV,
.init = NULL,
.create= grouping_create_stddev, // not an error, stddev calculates this too
.reset = grouping_reset_stddev, // not an error, stddev calculates this too
.free = grouping_free_stddev, // not an error, stddev calculates this too
.add = grouping_add_stddev, // not an error, stddev calculates this too
.flush = grouping_flush_coefficient_of_variation
},
{.name = "rsd", // alias of 'cv'
.hash = 0,
.value = RRDR_GROUPING_CV,
.init = NULL,
.create= grouping_create_stddev, // not an error, stddev calculates this too
.reset = grouping_reset_stddev, // not an error, stddev calculates this too
.free = grouping_free_stddev, // not an error, stddev calculates this too
.add = grouping_add_stddev, // not an error, stddev calculates this too
.flush = grouping_flush_coefficient_of_variation
},
/*
{.name = "mean", // same as average, no need to define it again
.hash = 0,
.value = RRDR_GROUPING_MEAN,
.setup = NULL,
.create= grouping_create_stddev,
.reset = grouping_reset_stddev,
.free = grouping_free_stddev,
.add = grouping_add_stddev,
.flush = grouping_flush_mean
},
*/
/*
{.name = "variance", // meaningless to offer
.hash = 0,
.value = RRDR_GROUPING_VARIANCE,
.setup = NULL,
.create= grouping_create_stddev,
.reset = grouping_reset_stddev,
.free = grouping_free_stddev,
.add = grouping_add_stddev,
.flush = grouping_flush_variance
},
*/
// single exponential smoothing
{.name = "ses",
.hash = 0,
.value = RRDR_GROUPING_SES,
.init = grouping_init_ses,
.create= grouping_create_ses,
.reset = grouping_reset_ses,
.free = grouping_free_ses,
.add = grouping_add_ses,
.flush = grouping_flush_ses
},
{.name = "ema", // alias for 'ses'
.hash = 0,
.value = RRDR_GROUPING_SES,
.init = NULL,
.create= grouping_create_ses,
.reset = grouping_reset_ses,
.free = grouping_free_ses,
.add = grouping_add_ses,
.flush = grouping_flush_ses
},
{.name = "ewma", // alias for ses
.hash = 0,
.value = RRDR_GROUPING_SES,
.init = NULL,
.create= grouping_create_ses,
.reset = grouping_reset_ses,
.free = grouping_free_ses,
.add = grouping_add_ses,
.flush = grouping_flush_ses
},
// double exponential smoothing
{.name = "des",
.hash = 0,
.value = RRDR_GROUPING_DES,
.init = grouping_init_des,
.create= grouping_create_des,
.reset = grouping_reset_des,
.free = grouping_free_des,
.add = grouping_add_des,
.flush = grouping_flush_des
},
// terminator
{.name = NULL,
.hash = 0,
.value = RRDR_GROUPING_UNDEFINED,
.init = NULL,
.create= grouping_create_average,
.reset = grouping_reset_average,
.free = grouping_free_average,
.add = grouping_add_average,
.flush = grouping_flush_average
}
};
void web_client_api_v1_init_grouping(void) {
int i;
for(i = 0; api_v1_data_groups[i].name ; i++) {
api_v1_data_groups[i].hash = simple_hash(api_v1_data_groups[i].name);
if(api_v1_data_groups[i].init)
api_v1_data_groups[i].init();
}
}
const char *group_method2string(RRDR_GROUPING group) {
int i;
for(i = 0; api_v1_data_groups[i].name ; i++) {
if(api_v1_data_groups[i].value == group) {
return api_v1_data_groups[i].name;
}
}
return "unknown-group-method";
}
RRDR_GROUPING web_client_api_request_v1_data_group(const char *name, RRDR_GROUPING def) {
int i;
uint32_t hash = simple_hash(name);
for(i = 0; api_v1_data_groups[i].name ; i++)
if(unlikely(hash == api_v1_data_groups[i].hash && !strcmp(name, api_v1_data_groups[i].name)))
return api_v1_data_groups[i].value;
return def;
}
// ----------------------------------------------------------------------------
static void rrdr_disable_not_selected_dimensions(RRDR *r, RRDR_OPTIONS options, const char *dims,
struct context_param *context_param_list)
{
RRDDIM *temp_rd = context_param_list ? context_param_list->rd : NULL;
int should_lock = (!context_param_list || !(context_param_list->flags & CONTEXT_FLAGS_ARCHIVE));
if (should_lock)
rrdset_check_rdlock(r->st);
if(unlikely(!dims || !*dims || (dims[0] == '*' && dims[1] == '\0'))) return;
int match_ids = 0, match_names = 0;
if(unlikely(options & RRDR_OPTION_MATCH_IDS))
match_ids = 1;
if(unlikely(options & RRDR_OPTION_MATCH_NAMES))
match_names = 1;
if(likely(!match_ids && !match_names))
match_ids = match_names = 1;
SIMPLE_PATTERN *pattern = simple_pattern_create(dims, ",|\t\r\n\f\v", SIMPLE_PATTERN_EXACT);
RRDDIM *d;
long c, dims_selected = 0, dims_not_hidden_not_zero = 0;
for(c = 0, d = temp_rd?temp_rd:r->st->dimensions; d ;c++, d = d->next) {
if( (match_ids && simple_pattern_matches(pattern, d->id))
|| (match_names && simple_pattern_matches(pattern, d->name))
) {
r->od[c] |= RRDR_DIMENSION_SELECTED;
if(unlikely(r->od[c] & RRDR_DIMENSION_HIDDEN)) r->od[c] &= ~RRDR_DIMENSION_HIDDEN;
dims_selected++;
// since the user needs this dimension
// make it appear as NONZERO, to return it
// even if the dimension has only zeros
// unless option non_zero is set
if(unlikely(!(options & RRDR_OPTION_NONZERO)))
r->od[c] |= RRDR_DIMENSION_NONZERO;
// count the visible dimensions
if(likely(r->od[c] & RRDR_DIMENSION_NONZERO))
dims_not_hidden_not_zero++;
}
else {
r->od[c] |= RRDR_DIMENSION_HIDDEN;
if(unlikely(r->od[c] & RRDR_DIMENSION_SELECTED)) r->od[c] &= ~RRDR_DIMENSION_SELECTED;
}
}
simple_pattern_free(pattern);
// check if all dimensions are hidden
if(unlikely(!dims_not_hidden_not_zero && dims_selected)) {
// there are a few selected dimensions
// but they are all zero
// enable the selected ones
// to avoid returning an empty chart
for(c = 0, d = temp_rd?temp_rd:r->st->dimensions; d ;c++, d = d->next)
if(unlikely(r->od[c] & RRDR_DIMENSION_SELECTED))
r->od[c] |= RRDR_DIMENSION_NONZERO;
}
}
// ----------------------------------------------------------------------------
// helpers to find our way in RRDR
static inline RRDR_VALUE_FLAGS *rrdr_line_options(RRDR *r, long rrdr_line) {
return &r->o[ rrdr_line * r->d ];
}
static inline calculated_number *rrdr_line_values(RRDR *r, long rrdr_line) {
return &r->v[ rrdr_line * r->d ];
}
static inline long rrdr_line_init(RRDR *r, time_t t, long rrdr_line) {
rrdr_line++;
#ifdef NETDATA_INTERNAL_CHECKS
if(unlikely(rrdr_line >= r->n))
error("INTERNAL ERROR: requested to step above RRDR size for chart '%s'", r->st->name);
if(unlikely(r->t[rrdr_line] != 0 && r->t[rrdr_line] != t))
error("INTERNAL ERROR: overwriting the timestamp of RRDR line %zu from %zu to %zu, of chart '%s'", (size_t)rrdr_line, (size_t)r->t[rrdr_line], (size_t)t, r->st->name);
#endif
// save the time
r->t[rrdr_line] = t;
return rrdr_line;
}
static inline void rrdr_done(RRDR *r, long rrdr_line) {
r->rows = rrdr_line + 1;
}
// ----------------------------------------------------------------------------
// fill RRDR for a single dimension
static inline void do_dimension_variablestep(
RRDR *r
, long points_wanted
, RRDDIM *rd
, long dim_id_in_rrdr
, time_t after_wanted
, time_t before_wanted
, uint32_t options
){
// RRDSET *st = r->st;
time_t
now = after_wanted,
dt = r->update_every,
max_date = 0,
min_date = 0;
long
// group_size = r->group,
points_added = 0,
values_in_group = 0,
values_in_group_non_zero = 0,
rrdr_line = -1;
RRDR_VALUE_FLAGS
group_value_flags = RRDR_VALUE_NOTHING;
struct rrddim_query_handle handle;
calculated_number min = r->min, max = r->max;
size_t db_points_read = 0;
time_t db_now = now;
storage_number n_curr, n_prev = SN_EMPTY_SLOT;
calculated_number value;
for(rd->state->query_ops.init(rd, &handle, now, before_wanted) ; points_added < points_wanted ; now += dt) {
// make sure we return data in the proper time range
if (unlikely(now > before_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
r->internal.log = "stopped, because attempted to access the db after 'wanted before'";
#endif
break;
}
if (unlikely(now < after_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
r->internal.log = "skipped, because attempted to access the db before 'wanted after'";
#endif
continue;
}
while (now >= db_now && (!rd->state->query_ops.is_finished(&handle) ||
does_storage_number_exist(n_prev))) {
value = NAN;
if (does_storage_number_exist(n_prev)) {
// use the previously read database value
n_curr = n_prev;
} else {
// read the value from the database
n_curr = rd->state->query_ops.next_metric(&handle, &db_now);
}
n_prev = SN_EMPTY_SLOT;
// db_now has a different value than above
if (likely(now >= db_now)) {
if (likely(does_storage_number_exist(n_curr))) {
if (options & RRDR_OPTION_ANOMALY_BIT)
value = (n_curr & SN_ANOMALY_BIT) ? 0.0 : 100.0;
else
value = unpack_storage_number(n_curr);
if (likely(value != 0.0))
values_in_group_non_zero++;
if (unlikely(did_storage_number_reset(n_curr)))
group_value_flags |= RRDR_VALUE_RESET;
}
} else {
// We must postpone processing the value and fill the result with gaps instead
if (likely(does_storage_number_exist(n_curr))) {
n_prev = n_curr;
}
}
// add this value to grouping
r->internal.grouping_add(r, value);
values_in_group++;
db_points_read++;
}
if (0 == values_in_group) {
// add NAN to grouping
r->internal.grouping_add(r, NAN);
}
rrdr_line = rrdr_line_init(r, now, rrdr_line);
if(unlikely(!min_date)) min_date = now;
max_date = now;
// find the place to store our values
RRDR_VALUE_FLAGS *rrdr_value_options_ptr = &r->o[rrdr_line * r->d + dim_id_in_rrdr];
// update the dimension options
if(likely(values_in_group_non_zero))
r->od[dim_id_in_rrdr] |= RRDR_DIMENSION_NONZERO;
// store the specific point options
*rrdr_value_options_ptr = group_value_flags;
// store the value
value = r->internal.grouping_flush(r, rrdr_value_options_ptr);
r->v[rrdr_line * r->d + dim_id_in_rrdr] = value;
if(likely(points_added || dim_id_in_rrdr)) {
// find the min/max across all dimensions
if(unlikely(value < min)) min = value;
if(unlikely(value > max)) max = value;
}
else {
// runs only when dim_id_in_rrdr == 0 && points_added == 0
// so, on the first point added for the query.
min = max = value;
}
points_added++;
values_in_group = 0;
group_value_flags = RRDR_VALUE_NOTHING;
values_in_group_non_zero = 0;
}
rd->state->query_ops.finalize(&handle);
r->internal.db_points_read += db_points_read;
r->internal.result_points_generated += points_added;
r->min = min;
r->max = max;
r->before = max_date;
r->after = min_date - (r->group - 1) * dt;
rrdr_done(r, rrdr_line);
#ifdef NETDATA_INTERNAL_CHECKS
if(unlikely(r->rows != points_added))
error("INTERNAL ERROR: %s.%s added %zu rows, but RRDR says I added %zu.", r->st->name, rd->name, (size_t)points_added, (size_t)r->rows);
#endif
}
static inline void do_dimension_fixedstep(
RRDR *r
, long points_wanted
, RRDDIM *rd
, long dim_id_in_rrdr
, time_t after_wanted
, time_t before_wanted
, uint32_t options
){
RRDSET *st = r->st;
time_t
now = after_wanted,
dt = r->update_every / r->group, /* usually is st->update_every */
max_date = 0,
min_date = 0;
long
group_size = r->group,
points_added = 0,
values_in_group = 0,
values_in_group_non_zero = 0,
rrdr_line = -1;
RRDR_VALUE_FLAGS
group_value_flags = RRDR_VALUE_NOTHING;
struct rrddim_query_handle handle;
calculated_number min = r->min, max = r->max;
size_t db_points_read = 0;
time_t db_now = now;
for(rd->state->query_ops.init(rd, &handle, now, before_wanted) ; points_added < points_wanted ; now += dt) {
// make sure we return data in the proper time range
if(unlikely(now > before_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
r->internal.log = "stopped, because attempted to access the db after 'wanted before'";
#endif
break;
}
if(unlikely(now < after_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
r->internal.log = "skipped, because attempted to access the db before 'wanted after'";
#endif
continue;
}
// read the value from the database
//storage_number n = rd->values[slot];
#ifdef NETDATA_INTERNAL_CHECKS
if ((rd->rrd_memory_mode != RRD_MEMORY_MODE_DBENGINE) &&
(rrdset_time2slot(st, now) != (long unsigned)handle.slotted.slot)) {
error("INTERNAL CHECK: Unaligned query for %s, database slot: %lu, expected slot: %lu", rd->id, (long unsigned)handle.slotted.slot, rrdset_time2slot(st, now));
}
#endif
db_now = now; // this is needed to set db_now in case the next_metric implementation does not set it
storage_number n = rd->state->query_ops.next_metric(&handle, &db_now);
if(unlikely(db_now > before_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
r->internal.log = "stopped, because attempted to access the db after 'wanted before'";
#endif
break;
}
for ( ; now <= db_now ; now += dt) {
calculated_number value = NAN;
if(likely(now >= db_now && does_storage_number_exist(n))) {
#if defined(NETDATA_INTERNAL_CHECKS) && defined(ENABLE_DBENGINE)
if ((rd->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) && (now != handle.rrdeng.now)) {
error("INTERNAL CHECK: Unaligned query for %s, database time: %ld, expected time: %ld", rd->id, (long)handle.rrdeng.now, (long)now);
}
#endif
if (options & RRDR_OPTION_ANOMALY_BIT)
value = (n & SN_ANOMALY_BIT) ? 0.0 : 100.0;
else
value = unpack_storage_number(n);
if(likely(value != 0.0))
values_in_group_non_zero++;
if(unlikely(did_storage_number_reset(n)))
group_value_flags |= RRDR_VALUE_RESET;
}
// add this value for grouping
r->internal.grouping_add(r, value);
values_in_group++;
db_points_read++;
if(unlikely(values_in_group == group_size)) {
rrdr_line = rrdr_line_init(r, now, rrdr_line);
if(unlikely(!min_date)) min_date = now;
max_date = now;
// find the place to store our values
RRDR_VALUE_FLAGS *rrdr_value_options_ptr = &r->o[rrdr_line * r->d + dim_id_in_rrdr];
// update the dimension options
if(likely(values_in_group_non_zero))
r->od[dim_id_in_rrdr] |= RRDR_DIMENSION_NONZERO;
// store the specific point options
*rrdr_value_options_ptr = group_value_flags;
// store the value
calculated_number value = r->internal.grouping_flush(r, rrdr_value_options_ptr);
r->v[rrdr_line * r->d + dim_id_in_rrdr] = value;
if(likely(points_added || dim_id_in_rrdr)) {
// find the min/max across all dimensions
if(unlikely(value < min)) min = value;
if(unlikely(value > max)) max = value;
}
else {
// runs only when dim_id_in_rrdr == 0 && points_added == 0
// so, on the first point added for the query.
min = max = value;
}
points_added++;
values_in_group = 0;
group_value_flags = RRDR_VALUE_NOTHING;
values_in_group_non_zero = 0;
}
}
now = db_now;
}
rd->state->query_ops.finalize(&handle);
r->internal.db_points_read += db_points_read;
r->internal.result_points_generated += points_added;
r->min = min;
r->max = max;
r->before = max_date;
r->after = min_date - (r->group - 1) * dt;
rrdr_done(r, rrdr_line);
#ifdef NETDATA_INTERNAL_CHECKS
if(unlikely(r->rows != points_added))
error("INTERNAL ERROR: %s.%s added %zu rows, but RRDR says I added %zu.", r->st->name, rd->name, (size_t)points_added, (size_t)r->rows);
#endif
}
// ----------------------------------------------------------------------------
// fill RRDR for the whole chart
#ifdef NETDATA_INTERNAL_CHECKS
static void rrd2rrdr_log_request_response_metadata(RRDR *r
, RRDR_GROUPING group_method
, int aligned
, long group
, long resampling_time
, long resampling_group
, time_t after_wanted
, time_t after_requested
, time_t before_wanted
, time_t before_requested
, long points_requested
, long points_wanted
//, size_t after_slot
//, size_t before_slot
, const char *msg
) {
netdata_rwlock_rdlock(&r->st->rrdset_rwlock);
info("INTERNAL ERROR: rrd2rrdr() on %s update every %d with %s grouping %s (group: %ld, resampling_time: %ld, resampling_group: %ld), "
"after (got: %zu, want: %zu, req: %zu, db: %zu), "
"before (got: %zu, want: %zu, req: %zu, db: %zu), "
"duration (got: %zu, want: %zu, req: %zu, db: %zu), "
//"slot (after: %zu, before: %zu, delta: %zu), "
"points (got: %ld, want: %ld, req: %ld, db: %ld), "
"%s"
, r->st->name
, r->st->update_every
// grouping
, (aligned) ? "aligned" : "unaligned"
, group_method2string(group_method)
, group
, resampling_time
, resampling_group
// after
, (size_t)r->after
, (size_t)after_wanted
, (size_t)after_requested
, (size_t)rrdset_first_entry_t_nolock(r->st)
// before
, (size_t)r->before
, (size_t)before_wanted
, (size_t)before_requested
, (size_t)rrdset_last_entry_t_nolock(r->st)
// duration
, (size_t)(r->before - r->after + r->st->update_every)
, (size_t)(before_wanted - after_wanted + r->st->update_every)
, (size_t)(before_requested - after_requested)
, (size_t)((rrdset_last_entry_t_nolock(r->st) - rrdset_first_entry_t_nolock(r->st)) + r->st->update_every)
// slot
/*
, after_slot
, before_slot
, (after_slot > before_slot) ? (r->st->entries - after_slot + before_slot) : (before_slot - after_slot)
*/
// points
, r->rows
, points_wanted
, points_requested
, r->st->entries
// message
, msg
);
netdata_rwlock_unlock(&r->st->rrdset_rwlock);
}
#endif // NETDATA_INTERNAL_CHECKS
// Returns 1 if an absolute period was requested or 0 if it was a relative period
static int rrdr_convert_before_after_to_absolute(
long long *after_requestedp
, long long *before_requestedp
, int update_every
, time_t first_entry_t
, time_t last_entry_t
, RRDR_OPTIONS options
) {
int absolute_period_requested = -1;
long long after_requested, before_requested;
before_requested = *before_requestedp;
after_requested = *after_requestedp;
if(before_requested == 0 && after_requested == 0) {
// dump the all the data
before_requested = last_entry_t;
after_requested = first_entry_t;
absolute_period_requested = 0;
}
// allow relative for before (smaller than API_RELATIVE_TIME_MAX)
if(ABS(before_requested) <= API_RELATIVE_TIME_MAX) {
if(ABS(before_requested) % update_every) {
// make sure it is multiple of st->update_every
if(before_requested < 0) before_requested = before_requested - update_every -
before_requested % update_every;
else before_requested = before_requested + update_every - before_requested % update_every;
}
if(before_requested > 0) before_requested = first_entry_t + before_requested;
else before_requested = last_entry_t + before_requested; //last_entry_t is not really now_t
//TODO: fix before_requested to be relative to now_t
absolute_period_requested = 0;
}
// allow relative for after (smaller than API_RELATIVE_TIME_MAX)
if(ABS(after_requested) <= API_RELATIVE_TIME_MAX) {
if(after_requested == 0) after_requested = -update_every;
if(ABS(after_requested) % update_every) {
// make sure it is multiple of st->update_every
if(after_requested < 0) after_requested = after_requested - update_every - after_requested % update_every;
else after_requested = after_requested + update_every - after_requested % update_every;
}
after_requested = before_requested + after_requested;
absolute_period_requested = 0;
}
if(absolute_period_requested == -1)
absolute_period_requested = 1;
// make sure they are within our timeframe
if(before_requested > last_entry_t) before_requested = last_entry_t;
if(before_requested < first_entry_t && !(options & RRDR_OPTION_ALLOW_PAST))
before_requested = first_entry_t;
if(after_requested > last_entry_t) after_requested = last_entry_t;
if(after_requested < first_entry_t && !(options & RRDR_OPTION_ALLOW_PAST))
after_requested = first_entry_t;
// check if they are reversed
if(after_requested > before_requested) {
time_t tmp = before_requested;
before_requested = after_requested;
after_requested = tmp;
}
*before_requestedp = before_requested;
*after_requestedp = after_requested;
return absolute_period_requested;
}
static RRDR *rrd2rrdr_fixedstep(
RRDSET *st
, long points_requested
, long long after_requested
, long long before_requested
, RRDR_GROUPING group_method
, long resampling_time_requested
, RRDR_OPTIONS options
, const char *dimensions
, int update_every
, time_t first_entry_t
, time_t last_entry_t
, int absolute_period_requested
, struct context_param *context_param_list
) {
int aligned = !(options & RRDR_OPTION_NOT_ALIGNED);
// the duration of the chart
time_t duration = before_requested - after_requested;
long available_points = duration / update_every;
RRDDIM *temp_rd = context_param_list ? context_param_list->rd : NULL;
if(duration <= 0 || available_points <= 0)
return rrdr_create(st, 1, context_param_list);
// check the number of wanted points in the result
if(unlikely(points_requested < 0)) points_requested = -points_requested;
if(unlikely(points_requested > available_points)) points_requested = available_points;
if(unlikely(points_requested == 0)) points_requested = available_points;
// calculate the desired grouping of source data points
long group = available_points / points_requested;
if(unlikely(group <= 0)) group = 1;
if(unlikely(available_points % points_requested > points_requested / 2)) group++; // rounding to the closest integer
// resampling_time_requested enforces a certain grouping multiple
calculated_number resampling_divisor = 1.0;
long resampling_group = 1;
if(unlikely(resampling_time_requested > update_every)) {
if (unlikely(resampling_time_requested > duration)) {
// group_time is above the available duration
#ifdef NETDATA_INTERNAL_CHECKS
info("INTERNAL CHECK: %s: requested gtime %ld secs, is greater than the desired duration %ld secs", st->id, resampling_time_requested, duration);
#endif
after_requested = before_requested - resampling_time_requested;
duration = before_requested - after_requested;
available_points = duration / update_every;
group = available_points / points_requested;
}
// if the duration is not aligned to resampling time
// extend the duration to the past, to avoid a gap at the chart
// only when the missing duration is above 1/10th of a point
if(duration % resampling_time_requested) {
time_t delta = duration % resampling_time_requested;
if(delta > resampling_time_requested / 10) {
after_requested -= resampling_time_requested - delta;
duration = before_requested - after_requested;
available_points = duration / update_every;
group = available_points / points_requested;
}
}
// the points we should group to satisfy gtime
resampling_group = resampling_time_requested / update_every;
if(unlikely(resampling_time_requested % update_every)) {
#ifdef NETDATA_INTERNAL_CHECKS
info("INTERNAL CHECK: %s: requested gtime %ld secs, is not a multiple of the chart's data collection frequency %d secs", st->id, resampling_time_requested, update_every);
#endif
resampling_group++;
}
// adapt group according to resampling_group
if(unlikely(group < resampling_group)) group = resampling_group; // do not allow grouping below the desired one
if(unlikely(group % resampling_group)) group += resampling_group - (group % resampling_group); // make sure group is multiple of resampling_group
//resampling_divisor = group / resampling_group;
resampling_divisor = (calculated_number)(group * update_every) / (calculated_number)resampling_time_requested;
}
// now that we have group,
// align the requested timeframe to fit it.
if(aligned) {
// alignment has been requested, so align the values
before_requested -= before_requested % (group * update_every);
after_requested -= after_requested % (group * update_every);
}
// we align the request on requested_before
time_t before_wanted = before_requested;
if(likely(before_wanted > last_entry_t)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, before_wanted is after db max", st->name);
#endif
before_wanted = last_entry_t - (last_entry_t % ( ((aligned)?group:1) * update_every ));
}
//size_t before_slot = rrdset_time2slot(st, before_wanted);
// we need to estimate the number of points, for having
// an integer number of values per point
long points_wanted = (before_wanted - after_requested) / (update_every * group);
time_t after_wanted = before_wanted - (points_wanted * group * update_every) + update_every;
if(unlikely(after_wanted < first_entry_t)) {
// hm... we go to the past, calculate again points_wanted using all the db from before_wanted to the beginning
points_wanted = (before_wanted - first_entry_t) / group;
// recalculate after wanted with the new number of points
after_wanted = before_wanted - (points_wanted * group * update_every) + update_every;
if(unlikely(after_wanted < first_entry_t)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, after_wanted is before db min", st->name);
#endif
after_wanted = first_entry_t - (first_entry_t % ( ((aligned)?group:1) * update_every )) + ( ((aligned)?group:1) * update_every );
}
}
//size_t after_slot = rrdset_time2slot(st, after_wanted);
// check if they are reversed
if(unlikely(after_wanted > before_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, reversed wanted after/before", st->name);
#endif
time_t tmp = before_wanted;
before_wanted = after_wanted;
after_wanted = tmp;
}
// recalculate points_wanted using the final time-frame
points_wanted = (before_wanted - after_wanted) / update_every / group + 1;
if(unlikely(points_wanted < 0)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, points_wanted is %ld", st->name, points_wanted);
#endif
points_wanted = 0;
}
#ifdef NETDATA_INTERNAL_CHECKS
duration = before_wanted - after_wanted;
if(after_wanted < first_entry_t)
error("INTERNAL CHECK: after_wanted %u is too small, minimum %u", (uint32_t)after_wanted, (uint32_t)first_entry_t);
if(after_wanted > last_entry_t)
error("INTERNAL CHECK: after_wanted %u is too big, maximum %u", (uint32_t)after_wanted, (uint32_t)last_entry_t);
if(before_wanted < first_entry_t)
error("INTERNAL CHECK: before_wanted %u is too small, minimum %u", (uint32_t)before_wanted, (uint32_t)first_entry_t);
if(before_wanted > last_entry_t)
error("INTERNAL CHECK: before_wanted %u is too big, maximum %u", (uint32_t)before_wanted, (uint32_t)last_entry_t);
/*
if(before_slot >= (size_t)st->entries)
error("INTERNAL CHECK: before_slot is invalid %zu, expected 0 to %ld", before_slot, st->entries - 1);
if(after_slot >= (size_t)st->entries)
error("INTERNAL CHECK: after_slot is invalid %zu, expected 0 to %ld", after_slot, st->entries - 1);
*/
if(points_wanted > (before_wanted - after_wanted) / group / update_every + 1)
error("INTERNAL CHECK: points_wanted %ld is more than points %ld", points_wanted, (before_wanted - after_wanted) / group / update_every + 1);
if(group < resampling_group)
error("INTERNAL CHECK: group %ld is less than the desired group points %ld", group, resampling_group);
if(group > resampling_group && group % resampling_group)
error("INTERNAL CHECK: group %ld is not a multiple of the desired group points %ld", group, resampling_group);
#endif
// -------------------------------------------------------------------------
// initialize our result set
// this also locks the chart for us
RRDR *r = rrdr_create(st, points_wanted, context_param_list);
if(unlikely(!r)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL CHECK: Cannot create RRDR for %s, after=%u, before=%u, duration=%u, points=%ld", st->id, (uint32_t)after_wanted, (uint32_t)before_wanted, (uint32_t)duration, points_wanted);
#endif
return NULL;
}
if(unlikely(!r->d || !points_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL CHECK: Returning empty RRDR (no dimensions in RRDSET) for %s, after=%u, before=%u, duration=%zu, points=%ld", st->id, (uint32_t)after_wanted, (uint32_t)before_wanted, (size_t)duration, points_wanted);
#endif
return r;
}
if(unlikely(absolute_period_requested == 1))
r->result_options |= RRDR_RESULT_OPTION_ABSOLUTE;
else
r->result_options |= RRDR_RESULT_OPTION_RELATIVE;
// find how many dimensions we have
long dimensions_count = r->d;
// -------------------------------------------------------------------------
// initialize RRDR
r->group = group;
r->update_every = (int)group * update_every;
r->before = before_wanted;
r->after = after_wanted;
r->internal.points_wanted = points_wanted;
r->internal.resampling_group = resampling_group;
r->internal.resampling_divisor = resampling_divisor;
// -------------------------------------------------------------------------
// assign the processor functions
{
int i, found = 0;
for(i = 0; !found && api_v1_data_groups[i].name ;i++) {
if(api_v1_data_groups[i].value == group_method) {
r->internal.grouping_create= api_v1_data_groups[i].create;
r->internal.grouping_reset = api_v1_data_groups[i].reset;
r->internal.grouping_free = api_v1_data_groups[i].free;
r->internal.grouping_add = api_v1_data_groups[i].add;
r->internal.grouping_flush = api_v1_data_groups[i].flush;
found = 1;
}
}
if(!found) {
errno = 0;
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: grouping method %u not found for chart '%s'. Using 'average'", (unsigned int)group_method, r->st->name);
#endif
r->internal.grouping_create= grouping_create_average;
r->internal.grouping_reset = grouping_reset_average;
r->internal.grouping_free = grouping_free_average;
r->internal.grouping_add = grouping_add_average;
r->internal.grouping_flush = grouping_flush_average;
}
}
// allocate any memory required by the grouping method
r->internal.grouping_data = r->internal.grouping_create(r);
// -------------------------------------------------------------------------
// disable the not-wanted dimensions
if (context_param_list && !(context_param_list->flags & CONTEXT_FLAGS_ARCHIVE))
rrdset_check_rdlock(st);
if(dimensions)
rrdr_disable_not_selected_dimensions(r, options, dimensions, context_param_list);
// -------------------------------------------------------------------------
// do the work for each dimension
time_t max_after = 0, min_before = 0;
long max_rows = 0;
RRDDIM *rd;
long c, dimensions_used = 0, dimensions_nonzero = 0;
for(rd = temp_rd?temp_rd:st->dimensions, c = 0 ; rd && c < dimensions_count ; rd = rd->next, c++) {
// if we need a percentage, we need to calculate all dimensions
if(unlikely(!(options & RRDR_OPTION_PERCENTAGE) && (r->od[c] & RRDR_DIMENSION_HIDDEN))) {
if(unlikely(r->od[c] & RRDR_DIMENSION_SELECTED)) r->od[c] &= ~RRDR_DIMENSION_SELECTED;
continue;
}
r->od[c] |= RRDR_DIMENSION_SELECTED;
// reset the grouping for the new dimension
r->internal.grouping_reset(r);
do_dimension_fixedstep(
r
, points_wanted
, rd
, c
, after_wanted
, before_wanted
, options
);
if(r->od[c] & RRDR_DIMENSION_NONZERO)
dimensions_nonzero++;
// verify all dimensions are aligned
if(unlikely(!dimensions_used)) {
min_before = r->before;
max_after = r->after;
max_rows = r->rows;
}
else {
if(r->after != max_after) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: 'after' mismatch between dimensions for chart '%s': max is %zu, dimension '%s' has %zu",
st->name, (size_t)max_after, rd->name, (size_t)r->after);
#endif
r->after = (r->after > max_after) ? r->after : max_after;
}
if(r->before != min_before) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: 'before' mismatch between dimensions for chart '%s': max is %zu, dimension '%s' has %zu",
st->name, (size_t)min_before, rd->name, (size_t)r->before);
#endif
r->before = (r->before < min_before) ? r->before : min_before;
}
if(r->rows != max_rows) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: 'rows' mismatch between dimensions for chart '%s': max is %zu, dimension '%s' has %zu",
st->name, (size_t)max_rows, rd->name, (size_t)r->rows);
#endif
r->rows = (r->rows > max_rows) ? r->rows : max_rows;
}
}
dimensions_used++;
}
#ifdef NETDATA_INTERNAL_CHECKS
if (dimensions_used) {
if(r->internal.log)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ r->internal.log);
if(r->rows != points_wanted)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "got 'points' is not wanted 'points'");
if(aligned && (r->before % group) != 0)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "'before' is not aligned but alignment is required");
// 'after' should not be aligned, since we start inside the first group
//if(aligned && (r->after % group) != 0)
// rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, after_slot, before_slot, "'after' is not aligned but alignment is required");
if(r->before != before_requested)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "chart is not aligned to requested 'before'");
if(r->before != before_wanted)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "got 'before' is not wanted 'before'");
// reported 'after' varies, depending on group
if(r->after != after_wanted)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "got 'after' is not wanted 'after'");
}
#endif
// free all resources used by the grouping method
r->internal.grouping_free(r);
// when all the dimensions are zero, we should return all of them
if(unlikely(options & RRDR_OPTION_NONZERO && !dimensions_nonzero)) {
// all the dimensions are zero
// mark them as NONZERO to send them all
for(rd = temp_rd?temp_rd:st->dimensions, c = 0 ; rd && c < dimensions_count ; rd = rd->next, c++) {
if(unlikely(r->od[c] & RRDR_DIMENSION_HIDDEN)) continue;
r->od[c] |= RRDR_DIMENSION_NONZERO;
}
}
rrdr_query_completed(r->internal.db_points_read, r->internal.result_points_generated);
return r;
}
#ifdef ENABLE_DBENGINE
static RRDR *rrd2rrdr_variablestep(
RRDSET *st
, long points_requested
, long long after_requested
, long long before_requested
, RRDR_GROUPING group_method
, long resampling_time_requested
, RRDR_OPTIONS options
, const char *dimensions
, int update_every
, time_t first_entry_t
, time_t last_entry_t
, int absolute_period_requested
, struct rrdeng_region_info *region_info_array
, struct context_param *context_param_list
) {
int aligned = !(options & RRDR_OPTION_NOT_ALIGNED);
// the duration of the chart
time_t duration = before_requested - after_requested;
long available_points = duration / update_every;
RRDDIM *temp_rd = context_param_list ? context_param_list->rd : NULL;
if(duration <= 0 || available_points <= 0) {
freez(region_info_array);
return rrdr_create(st, 1, context_param_list);
}
// check the number of wanted points in the result
if(unlikely(points_requested < 0)) points_requested = -points_requested;
if(unlikely(points_requested > available_points)) points_requested = available_points;
if(unlikely(points_requested == 0)) points_requested = available_points;
// calculate the desired grouping of source data points
long group = available_points / points_requested;
if(unlikely(group <= 0)) group = 1;
if(unlikely(available_points % points_requested > points_requested / 2)) group++; // rounding to the closest integer
// resampling_time_requested enforces a certain grouping multiple
calculated_number resampling_divisor = 1.0;
long resampling_group = 1;
if(unlikely(resampling_time_requested > update_every)) {
if (unlikely(resampling_time_requested > duration)) {
// group_time is above the available duration
#ifdef NETDATA_INTERNAL_CHECKS
info("INTERNAL CHECK: %s: requested gtime %ld secs, is greater than the desired duration %ld secs", st->id, resampling_time_requested, duration);
#endif
after_requested = before_requested - resampling_time_requested;
duration = before_requested - after_requested;
available_points = duration / update_every;
group = available_points / points_requested;
}
// if the duration is not aligned to resampling time
// extend the duration to the past, to avoid a gap at the chart
// only when the missing duration is above 1/10th of a point
if(duration % resampling_time_requested) {
time_t delta = duration % resampling_time_requested;
if(delta > resampling_time_requested / 10) {
after_requested -= resampling_time_requested - delta;
duration = before_requested - after_requested;
available_points = duration / update_every;
group = available_points / points_requested;
}
}
// the points we should group to satisfy gtime
resampling_group = resampling_time_requested / update_every;
if(unlikely(resampling_time_requested % update_every)) {
#ifdef NETDATA_INTERNAL_CHECKS
info("INTERNAL CHECK: %s: requested gtime %ld secs, is not a multiple of the chart's data collection frequency %d secs", st->id, resampling_time_requested, update_every);
#endif
resampling_group++;
}
// adapt group according to resampling_group
if(unlikely(group < resampling_group)) group = resampling_group; // do not allow grouping below the desired one
if(unlikely(group % resampling_group)) group += resampling_group - (group % resampling_group); // make sure group is multiple of resampling_group
//resampling_divisor = group / resampling_group;
resampling_divisor = (calculated_number)(group * update_every) / (calculated_number)resampling_time_requested;
}
// now that we have group,
// align the requested timeframe to fit it.
if(aligned) {
// alignment has been requested, so align the values
before_requested -= before_requested % (group * update_every);
after_requested -= after_requested % (group * update_every);
}
// we align the request on requested_before
time_t before_wanted = before_requested;
if(likely(before_wanted > last_entry_t)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, before_wanted is after db max", st->name);
#endif
before_wanted = last_entry_t - (last_entry_t % ( ((aligned)?group:1) * update_every ));
}
//size_t before_slot = rrdset_time2slot(st, before_wanted);
// we need to estimate the number of points, for having
// an integer number of values per point
long points_wanted = (before_wanted - after_requested) / (update_every * group);
time_t after_wanted = before_wanted - (points_wanted * group * update_every) + update_every;
if(unlikely(after_wanted < first_entry_t)) {
// hm... we go to the past, calculate again points_wanted using all the db from before_wanted to the beginning
points_wanted = (before_wanted - first_entry_t) / group;
// recalculate after wanted with the new number of points
after_wanted = before_wanted - (points_wanted * group * update_every) + update_every;
if(unlikely(after_wanted < first_entry_t)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, after_wanted is before db min", st->name);
#endif
after_wanted = first_entry_t - (first_entry_t % ( ((aligned)?group:1) * update_every )) + ( ((aligned)?group:1) * update_every );
}
}
//size_t after_slot = rrdset_time2slot(st, after_wanted);
// check if they are reversed
if(unlikely(after_wanted > before_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, reversed wanted after/before", st->name);
#endif
time_t tmp = before_wanted;
before_wanted = after_wanted;
after_wanted = tmp;
}
// recalculate points_wanted using the final time-frame
points_wanted = (before_wanted - after_wanted) / update_every / group + 1;
if(unlikely(points_wanted < 0)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: rrd2rrdr() on %s, points_wanted is %ld", st->name, points_wanted);
#endif
points_wanted = 0;
}
#ifdef NETDATA_INTERNAL_CHECKS
duration = before_wanted - after_wanted;
if(after_wanted < first_entry_t)
error("INTERNAL CHECK: after_wanted %u is too small, minimum %u", (uint32_t)after_wanted, (uint32_t)first_entry_t);
if(after_wanted > last_entry_t)
error("INTERNAL CHECK: after_wanted %u is too big, maximum %u", (uint32_t)after_wanted, (uint32_t)last_entry_t);
if(before_wanted < first_entry_t)
error("INTERNAL CHECK: before_wanted %u is too small, minimum %u", (uint32_t)before_wanted, (uint32_t)first_entry_t);
if(before_wanted > last_entry_t)
error("INTERNAL CHECK: before_wanted %u is too big, maximum %u", (uint32_t)before_wanted, (uint32_t)last_entry_t);
/*
if(before_slot >= (size_t)st->entries)
error("INTERNAL CHECK: before_slot is invalid %zu, expected 0 to %ld", before_slot, st->entries - 1);
if(after_slot >= (size_t)st->entries)
error("INTERNAL CHECK: after_slot is invalid %zu, expected 0 to %ld", after_slot, st->entries - 1);
*/
if(points_wanted > (before_wanted - after_wanted) / group / update_every + 1)
error("INTERNAL CHECK: points_wanted %ld is more than points %ld", points_wanted, (before_wanted - after_wanted) / group / update_every + 1);
if(group < resampling_group)
error("INTERNAL CHECK: group %ld is less than the desired group points %ld", group, resampling_group);
if(group > resampling_group && group % resampling_group)
error("INTERNAL CHECK: group %ld is not a multiple of the desired group points %ld", group, resampling_group);
#endif
// -------------------------------------------------------------------------
// initialize our result set
// this also locks the chart for us
RRDR *r = rrdr_create(st, points_wanted, context_param_list);
if(unlikely(!r)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL CHECK: Cannot create RRDR for %s, after=%u, before=%u, duration=%u, points=%ld", st->id, (uint32_t)after_wanted, (uint32_t)before_wanted, (uint32_t)duration, points_wanted);
#endif
freez(region_info_array);
return NULL;
}
if(unlikely(!r->d || !points_wanted)) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL CHECK: Returning empty RRDR (no dimensions in RRDSET) for %s, after=%u, before=%u, duration=%zu, points=%ld", st->id, (uint32_t)after_wanted, (uint32_t)before_wanted, (size_t)duration, points_wanted);
#endif
freez(region_info_array);
return r;
}
r->result_options |= RRDR_RESULT_OPTION_VARIABLE_STEP;
if(unlikely(absolute_period_requested == 1))
r->result_options |= RRDR_RESULT_OPTION_ABSOLUTE;
else
r->result_options |= RRDR_RESULT_OPTION_RELATIVE;
// find how many dimensions we have
long dimensions_count = r->d;
// -------------------------------------------------------------------------
// initialize RRDR
r->group = group;
r->update_every = (int)group * update_every;
r->before = before_wanted;
r->after = after_wanted;
r->internal.points_wanted = points_wanted;
r->internal.resampling_group = resampling_group;
r->internal.resampling_divisor = resampling_divisor;
// -------------------------------------------------------------------------
// assign the processor functions
{
int i, found = 0;
for(i = 0; !found && api_v1_data_groups[i].name ;i++) {
if(api_v1_data_groups[i].value == group_method) {
r->internal.grouping_create= api_v1_data_groups[i].create;
r->internal.grouping_reset = api_v1_data_groups[i].reset;
r->internal.grouping_free = api_v1_data_groups[i].free;
r->internal.grouping_add = api_v1_data_groups[i].add;
r->internal.grouping_flush = api_v1_data_groups[i].flush;
found = 1;
}
}
if(!found) {
errno = 0;
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: grouping method %u not found for chart '%s'. Using 'average'", (unsigned int)group_method, r->st->name);
#endif
r->internal.grouping_create= grouping_create_average;
r->internal.grouping_reset = grouping_reset_average;
r->internal.grouping_free = grouping_free_average;
r->internal.grouping_add = grouping_add_average;
r->internal.grouping_flush = grouping_flush_average;
}
}
// allocate any memory required by the grouping method
r->internal.grouping_data = r->internal.grouping_create(r);
// -------------------------------------------------------------------------
// disable the not-wanted dimensions
if (context_param_list && !(context_param_list->flags & CONTEXT_FLAGS_ARCHIVE))
rrdset_check_rdlock(st);
if(dimensions)
rrdr_disable_not_selected_dimensions(r, options, dimensions, context_param_list);
// -------------------------------------------------------------------------
// do the work for each dimension
time_t max_after = 0, min_before = 0;
long max_rows = 0;
RRDDIM *rd;
long c, dimensions_used = 0, dimensions_nonzero = 0;
for(rd = temp_rd?temp_rd:st->dimensions, c = 0 ; rd && c < dimensions_count ; rd = rd->next, c++) {
// if we need a percentage, we need to calculate all dimensions
if(unlikely(!(options & RRDR_OPTION_PERCENTAGE) && (r->od[c] & RRDR_DIMENSION_HIDDEN))) {
if(unlikely(r->od[c] & RRDR_DIMENSION_SELECTED)) r->od[c] &= ~RRDR_DIMENSION_SELECTED;
continue;
}
r->od[c] |= RRDR_DIMENSION_SELECTED;
// reset the grouping for the new dimension
r->internal.grouping_reset(r);
do_dimension_variablestep(
r
, points_wanted
, rd
, c
, after_wanted
, before_wanted
, options
);
if(r->od[c] & RRDR_DIMENSION_NONZERO)
dimensions_nonzero++;
// verify all dimensions are aligned
if(unlikely(!dimensions_used)) {
min_before = r->before;
max_after = r->after;
max_rows = r->rows;
}
else {
if(r->after != max_after) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: 'after' mismatch between dimensions for chart '%s': max is %zu, dimension '%s' has %zu",
st->name, (size_t)max_after, rd->name, (size_t)r->after);
#endif
r->after = (r->after > max_after) ? r->after : max_after;
}
if(r->before != min_before) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: 'before' mismatch between dimensions for chart '%s': max is %zu, dimension '%s' has %zu",
st->name, (size_t)min_before, rd->name, (size_t)r->before);
#endif
r->before = (r->before < min_before) ? r->before : min_before;
}
if(r->rows != max_rows) {
#ifdef NETDATA_INTERNAL_CHECKS
error("INTERNAL ERROR: 'rows' mismatch between dimensions for chart '%s': max is %zu, dimension '%s' has %zu",
st->name, (size_t)max_rows, rd->name, (size_t)r->rows);
#endif
r->rows = (r->rows > max_rows) ? r->rows : max_rows;
}
}
dimensions_used++;
}
#ifdef NETDATA_INTERNAL_CHECKS
if (dimensions_used) {
if(r->internal.log)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ r->internal.log);
if(r->rows != points_wanted)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "got 'points' is not wanted 'points'");
if(aligned && (r->before % group) != 0)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "'before' is not aligned but alignment is required");
// 'after' should not be aligned, since we start inside the first group
//if(aligned && (r->after % group) != 0)
// rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, after_slot, before_slot, "'after' is not aligned but alignment is required");
if(r->before != before_requested)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "chart is not aligned to requested 'before'");
if(r->before != before_wanted)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "got 'before' is not wanted 'before'");
// reported 'after' varies, depending on group
if(r->after != after_wanted)
rrd2rrdr_log_request_response_metadata(r, group_method, aligned, group, resampling_time_requested, resampling_group, after_wanted, after_requested, before_wanted, before_requested, points_requested, points_wanted, /*after_slot, before_slot,*/ "got 'after' is not wanted 'after'");
}
#endif
// free all resources used by the grouping method
r->internal.grouping_free(r);
// when all the dimensions are zero, we should return all of them
if(unlikely(options & RRDR_OPTION_NONZERO && !dimensions_nonzero)) {
// all the dimensions are zero
// mark them as NONZERO to send them all
for(rd = temp_rd?temp_rd:st->dimensions, c = 0 ; rd && c < dimensions_count ; rd = rd->next, c++) {
if(unlikely(r->od[c] & RRDR_DIMENSION_HIDDEN)) continue;
r->od[c] |= RRDR_DIMENSION_NONZERO;
}
}
rrdr_query_completed(r->internal.db_points_read, r->internal.result_points_generated);
freez(region_info_array);
return r;
}
#endif //#ifdef ENABLE_DBENGINE
RRDR *rrd2rrdr(
RRDSET *st
, long points_requested
, long long after_requested
, long long before_requested
, RRDR_GROUPING group_method
, long resampling_time_requested
, RRDR_OPTIONS options
, const char *dimensions
, struct context_param *context_param_list
)
{
int rrd_update_every;
int absolute_period_requested;
time_t first_entry_t;
time_t last_entry_t;
if (context_param_list) {
first_entry_t = context_param_list->first_entry_t;
last_entry_t = context_param_list->last_entry_t;
} else {
rrdset_rdlock(st);
first_entry_t = rrdset_first_entry_t_nolock(st);
last_entry_t = rrdset_last_entry_t_nolock(st);
rrdset_unlock(st);
}
rrd_update_every = st->update_every;
absolute_period_requested = rrdr_convert_before_after_to_absolute(&after_requested, &before_requested,
rrd_update_every, first_entry_t,
last_entry_t, options);
if (options & RRDR_OPTION_ALLOW_PAST)
if (first_entry_t > after_requested)
first_entry_t = after_requested;
if (context_param_list && !(context_param_list->flags & CONTEXT_FLAGS_ARCHIVE)) {
rebuild_context_param_list(context_param_list, after_requested);
st = context_param_list->rd ? context_param_list->rd->rrdset : NULL;
if (unlikely(!st))
return NULL;
}
#ifdef ENABLE_DBENGINE
if (st->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
struct rrdeng_region_info *region_info_array;
unsigned regions, max_interval;
/* This call takes the chart read-lock */
regions = rrdeng_variable_step_boundaries(st, after_requested, before_requested,
&region_info_array, &max_interval, context_param_list);
if (1 == regions) {
if (region_info_array) {
if (rrd_update_every != region_info_array[0].update_every) {
rrd_update_every = region_info_array[0].update_every;
/* recalculate query alignment */
absolute_period_requested =
rrdr_convert_before_after_to_absolute(&after_requested, &before_requested, rrd_update_every,
first_entry_t, last_entry_t, options);
}
freez(region_info_array);
}
return rrd2rrdr_fixedstep(st, points_requested, after_requested, before_requested, group_method,
resampling_time_requested, options, dimensions, rrd_update_every,
first_entry_t, last_entry_t, absolute_period_requested, context_param_list);
} else {
if (rrd_update_every != (uint16_t)max_interval) {
rrd_update_every = (uint16_t) max_interval;
/* recalculate query alignment */
absolute_period_requested = rrdr_convert_before_after_to_absolute(&after_requested, &before_requested,
rrd_update_every, first_entry_t,
last_entry_t, options);
}
return rrd2rrdr_variablestep(st, points_requested, after_requested, before_requested, group_method,
resampling_time_requested, options, dimensions, rrd_update_every,
first_entry_t, last_entry_t, absolute_period_requested, region_info_array, context_param_list);
}
}
#endif
return rrd2rrdr_fixedstep(st, points_requested, after_requested, before_requested, group_method,
resampling_time_requested, options, dimensions,
rrd_update_every, first_entry_t, last_entry_t, absolute_period_requested, context_param_list);
}
Reference in a new issue View git blame Copy permalink