PVE/Status/InfluxDB.pm

   1 package PVE::Status::InfluxDB;
   2
   3 use strict;
   4 use warnings;
   5
   6 use POSIX qw(isnan isinf);
   7 use Scalar::Util 'looks_like_number';
   8 use IO::Socket::IP;
   9 use LWP::UserAgent;
  10 use HTTP::Request;
  11
  12 use PVE::SafeSyslog;
  13
  14 use PVE::Status::Plugin;
  15
  16 use base('PVE::Status::Plugin');
  17
  18 sub type {
  19     return 'influxdb';
  20 }
  21
  22 sub properties {
  23     return {
  24         organization => {
  25             description => "The InfluxDB organization. Only necessary when using the http v2 api."
  26                ." Has no meaning when using v2 compatibility api.",
  27             type => 'string',
  28             optional => 1,
  29         },
  30         bucket => {
  31             description => "The InfluxDB bucket/db. Only necessary when using the http v2 api.",
  32             type => 'string',
  33             optional => 1,
  34         },
  35         token => {
  36             description => "The InfluxDB access token. Only necessary when using the http v2 api."
  37                 ." If the v2 compatibility api is used, use 'user:password' instead.",
  38             type => 'string',
  39             optional => 1,
  40         },
  41         'api-path-prefix' => {
  42             description => "An API path prefix inserted between '<host>:<port>/' and '/api2/'."
  43                 ." Can be useful if the InfluxDB service runs behind a reverse proxy.",
  44             type => 'string',
  45             optional => 1,
  46         },
  47         influxdbproto => {
  48             type => 'string',
  49             enum => ['udp', 'http', 'https'],
  50             default => 'udp',
  51             optional => 1,
  52         },
  53         'max-body-size' => {
  54             description => "InfluxDB max-body-size in bytes. Requests are batched up to this size.",
  55             type => 'integer',
  56             minimum => 1,
  57             default => 25_000_000,
  58         },
  59         'verify-certificate' => {
  60             description => "Set to 0 to disable certificate verification for https endpoints.",
  61             type => 'boolean',
  62             optional => 1,
  63             default => 1,
  64         },
  65     };
  66 }
  67 sub options {
  68     return {
  69         server => {},
  70         port => {},
  71         mtu => { optional => 1 },
  72         disable => { optional => 1 },
  73         organization => { optional => 1},
  74         bucket => { optional => 1},
  75         token => { optional => 1},
  76         influxdbproto => { optional => 1},
  77         timeout => { optional => 1},
  78         'max-body-size' => { optional => 1 },
  79         'api-path-prefix' => { optional => 1 },
  80         'verify-certificate' => { optional => 1 },
  81    };
  82 }
  83
  84 my $set_ssl_opts = sub {
  85     my ($cfg, $ua) = @_;
  86
  87     my $cert_verify = $cfg->{'verify-certificate'} // 1;
  88     if (!$cert_verify) {
  89         $ua->ssl_opts(
  90             verify_hostname => 0,
  91             SSL_verify_mode => IO::Socket::SSL::SSL_VERIFY_NONE,
  92         );
  93     }
  94
  95     return;
  96 };
  97
  98 # Plugin implementation
  99 sub update_node_status {
 100     my ($class, $txn, $node, $data, $ctime) = @_;
 101
 102     $ctime *= 1000000000;
 103
 104     build_influxdb_payload($class, $txn, $data, $ctime, "object=nodes,host=$node");
 105 }
 106
 107 sub update_qemu_status {
 108     my ($class, $txn, $vmid, $data, $ctime, $nodename) = @_;
 109
 110     $ctime *= 1000000000;
 111
 112     my $object = "object=qemu,vmid=$vmid,nodename=$nodename";
 113     if($data->{name} && $data->{name} ne '') {
 114         $object .= ",host=$data->{name}";
 115     }
 116     $object =~ s/\s/\\ /g;
 117
 118     # VMID is already added in base $object above, so exclude it from being re-added
 119     build_influxdb_payload($class, $txn, $data, $ctime, $object, { 'vmid' => 1 });
 120 }
 121
 122 sub update_lxc_status {
 123     my ($class, $txn, $vmid, $data, $ctime, $nodename) = @_;
 124
 125     $ctime *= 1000000000;
 126
 127     my $object = "object=lxc,vmid=$vmid,nodename=$nodename";
 128     if($data->{name} && $data->{name} ne '') {
 129         $object .= ",host=$data->{name}";
 130     }
 131     $object =~ s/\s/\\ /g;
 132
 133     # VMID is already added in base $object above, so exclude it from being re-added
 134     build_influxdb_payload($class, $txn, $data, $ctime, $object, { 'vmid' => 1 });
 135 }
 136
 137 sub update_storage_status {
 138     my ($class, $txn, $nodename, $storeid, $data, $ctime) = @_;
 139
 140     $ctime *= 1000000000;
 141
 142     my $object = "object=storages,nodename=$nodename,host=$storeid";
 143     if($data->{type} && $data->{type} ne '') {
 144         $object .= ",type=$data->{type}";
 145     }
 146     $object =~ s/\s/\\ /g;
 147
 148     build_influxdb_payload($class, $txn, $data, $ctime, $object);
 149 }
 150
 151 sub _send_batch_size {
 152     my ($class, $cfg) = @_;
 153     my $proto = $cfg->{influxdbproto} // 'udp';
 154     if ($proto ne 'udp') {
 155         return $cfg->{'max-body-size'} // 25_000_000;
 156     }
 157
 158     return $class->SUPER::_send_batch_size($cfg);
 159 }
 160
 161 sub send {
 162     my ($class, $connection, $data, $cfg) = @_;
 163
 164     my $proto = $cfg->{influxdbproto} // 'udp';
 165     if ($proto eq 'udp') {
 166         return $class->SUPER::send($connection, $data, $cfg);
 167     } elsif ($proto =~ m/^https?$/) {
 168         my $ua = LWP::UserAgent->new();
 169         $set_ssl_opts->($cfg, $ua);
 170         $ua->timeout($cfg->{timeout} // 1);
 171         $connection->content($data);
 172         my $response = $ua->request($connection);
 173
 174         if (!$response->is_success) {
 175             my $err = $response->status_line;
 176             die "$err\n";
 177         }
 178     } else {
 179         die "invalid protocol\n";
 180     }
 181
 182     return;
 183 }
 184
 185 sub _disconnect {
 186     my ($class, $connection, $cfg) = @_;
 187     my $proto = $cfg->{influxdbproto} // 'udp';
 188     if ($proto eq 'udp') {
 189         return $class->SUPER::_disconnect($connection, $cfg);
 190     }
 191
 192     return;
 193 }
 194
 195 sub _get_v2url {
 196     my ($cfg, $api_path) = @_;
 197     my ($proto, $host, $port) = $cfg->@{qw(influxdbproto server port)};
 198     my $api_prefix = $cfg->{'api-path-prefix'} // '/';
 199     if ($api_prefix ne '/' && $api_prefix =~ m!^/*(.+)/*$!) {
 200         $api_prefix = "/$1/";
 201     }
 202
 203     if ($api_path ne 'health') {
 204         $api_path = "api/v2/${api_path}";
 205     }
 206
 207     return "${proto}://${host}:${port}${api_prefix}${api_path}";
 208 }
 209
 210 sub _connect {
 211     my ($class, $cfg, $id) = @_;
 212
 213     my $host = $cfg->{server};
 214     my $port = $cfg->{port};
 215     my $proto = $cfg->{influxdbproto} // 'udp';
 216
 217     if ($proto eq 'udp') {
 218         my $socket = IO::Socket::IP->new(
 219             PeerAddr    => $host,
 220             PeerPort    => $port,
 221             Proto       => 'udp',
 222         ) || die "couldn't create influxdb socket [$host]:$port - $@\n";
 223
 224         $socket->blocking(0);
 225
 226         return $socket;
 227     } elsif ($proto =~ m/^https?$/) {
 228         my $token = get_credentials($id, 1);
 229         my $org = $cfg->{organization} // 'proxmox';
 230         my $bucket = $cfg->{bucket} // 'proxmox';
 231         my $url = _get_v2url($cfg, "write?org=${org}&bucket=${bucket}");
 232
 233         my $req = HTTP::Request->new(POST => $url);
 234         if (defined($token)) {
 235             $req->header( "Authorization", "Token $token");
 236         }
 237
 238         return $req;
 239     }
 240
 241     die "cannot connect to InfluxDB: invalid protocol '$proto'\n";
 242 }
 243
 244 sub test_connection {
 245     my ($class, $cfg, $id) = @_;
 246
 247     # do not check connection for disabled plugins
 248     return if $cfg->{disable};
 249
 250     my $proto = $cfg->{influxdbproto} // 'udp';
 251     if ($proto eq 'udp') {
 252         return $class->SUPER::test_connection($cfg, $id);
 253     } elsif ($proto =~ m/^https?$/) {
 254         my $url = _get_v2url($cfg, "health");
 255         my $ua = LWP::UserAgent->new();
 256         $set_ssl_opts->($cfg, $ua);
 257         $ua->timeout($cfg->{timeout} // 1);
 258         # in the initial add connection test, the token may still be in $cfg
 259         my $token = $cfg->{token} // get_credentials($id, 1);
 260         if (defined($token)) {
 261             $ua->default_header("Authorization" => "Token $token");
 262         }
 263         my $response = $ua->get($url);
 264
 265         if (!$response->is_success) {
 266             my $err = $response->status_line;
 267             die "$err\n";
 268         }
 269     } else {
 270         die "invalid protocol\n";
 271     }
 272
 273     return;
 274 }
 275
 276 sub build_influxdb_payload {
 277     my ($class, $txn, $data, $ctime, $tags, $excluded, $measurement, $instance) = @_;
 278
 279     # 'abc' and '123' are both valid hostnames, that confuses influx's type detection
 280     my $to_quote = { name => 1 };
 281
 282     my @values = ();
 283
 284     foreach my $key (sort keys %$data) {
 285         next if defined($excluded) && $excluded->{$key};
 286         my $value = $data->{$key};
 287         next if !defined($value);
 288
 289         if (!ref($value) && $value ne '') {
 290             # value is scalar
 291
 292             if (defined(my $v = prepare_value($value, $to_quote->{$key}))) {
 293                 push @values, "$key=$v";
 294             }
 295         } elsif (ref($value) eq 'HASH') {
 296             # value is a hash
 297
 298             if (!defined($measurement)) {
 299                 build_influxdb_payload($class, $txn, $value, $ctime, $tags, $excluded, $key);
 300             } elsif(!defined($instance)) {
 301                 build_influxdb_payload($class, $txn, $value, $ctime, $tags, $excluded, $measurement, $key);
 302             } else {
 303                 push @values, get_recursive_values($value);
 304             }
 305         }
 306     }
 307
 308     if (@values > 0) {
 309         my $mm = $measurement // 'system';
 310         my $tagstring = $tags;
 311         $tagstring .= ",instance=$instance" if defined($instance);
 312         my $valuestr = join(',', @values);
 313         $class->add_metric_data($txn, "$mm,$tagstring $valuestr $ctime\n");
 314     }
 315 }
 316
 317 sub get_recursive_values {
 318     my ($hash) = @_;
 319
 320     my @values = ();
 321
 322     foreach my $key (keys %$hash) {
 323         my $value = $hash->{$key};
 324         if(ref($value) eq 'HASH') {
 325             push(@values, get_recursive_values($value));
 326         } elsif (!ref($value) && $value ne '') {
 327             if (defined(my $v = prepare_value($value))) {
 328                 push @values, "$key=$v";
 329             }
 330         }
 331     }
 332
 333     return @values;
 334 }
 335
 336 sub prepare_value {
 337     my ($value, $force_quote) = @_;
 338
 339     # don't treat value like a number if quote is 1
 340     if (!$force_quote && looks_like_number($value)) {
 341         if (isnan($value) || isinf($value)) {
 342             # we cannot send influxdb NaN or Inf
 343             return undef;
 344         }
 345
 346         # influxdb also accepts 1.0e+10, etc.
 347         return $value;
 348     }
 349
 350     # non-numeric values require to be quoted, so escape " with \"
 351     $value =~ s/\"/\\\"/g;
 352     $value = "\"$value\"";
 353
 354     return $value;
 355 }
 356
 357 my $priv_dir = "/etc/pve/priv/metricserver";
 358
 359 sub cred_file_name {
 360     my ($id) = @_;
 361     return "${priv_dir}/${id}.pw";
 362 }
 363
 364 sub delete_credentials {
 365     my ($id) = @_;
 366
 367     if (my $cred_file = cred_file_name($id)) {
 368         unlink($cred_file)
 369             or warn "removing influxdb credentials file '$cred_file' failed: $!\n";
 370     }
 371
 372     return;
 373 }
 374
 375 sub set_credentials {
 376     my ($id, $token) = @_;
 377
 378     my $cred_file = cred_file_name($id);
 379
 380     mkdir $priv_dir;
 381
 382     PVE::Tools::file_set_contents($cred_file, "$token");
 383 }
 384
 385 sub get_credentials {
 386     my ($id, $silent) = @_;
 387
 388     my $cred_file = cred_file_name($id);
 389
 390     my $creds = eval { PVE::Tools::file_get_contents($cred_file) };
 391     warn "could not load credentials for '$id': $@\n" if $@ && !$silent;
 392
 393     return $creds;
 394 }
 395
 396 sub on_add_hook {
 397     my ($class, $id, $opts, $sensitive_opts) = @_;
 398
 399     my $token = $sensitive_opts->{token};
 400
 401     if (defined($token)) {
 402         set_credentials($id, $token);
 403     } else {
 404         delete_credentials($id);
 405     }
 406
 407     return undef;
 408 }
 409
 410 sub on_update_hook {
 411     my ($class, $id, $opts, $sensitive_opts) = @_;
 412     return if !exists($sensitive_opts->{token});
 413
 414     my $token = $sensitive_opts->{token};
 415     if (defined($token)) {
 416         set_credentials($id, $token);
 417     } else {
 418         delete_credentials($id);
 419     }
 420
 421     return undef;
 422 }
 423
 424 sub on_delete_hook {
 425     my ($class, $id, $opts) = @_;
 426
 427     delete_credentials($id);
 428
 429     return undef;
 430 }
 431
 432
 433 1;