<?php
    /*******************************************************************************
     *
     * File:         $Id$
     * Modified_On:  $Date$
     * Modified_By:  $Author$
     * License:      Commercial
     * Copyright:    Copyright 2009-2020 by Urban-Software.de / Thomas Urban
     *******************************************************************************/

    // autoload dependencies ( namely InfluxDB client )
    require_once __DIR__ . '/../vendor/autoload.php';

    use Elastic\Elasticsearch\Exception\ClientResponseException;
    use Elastic\Elasticsearch\Exception\ServerResponseException;
    use Elastic\Elasticsearch;
    use Elastic\Elasticsearch\ClientBuilder;

// UW9ZM2RZY0JDSk5YNzR2YkpvVnI6OHJoeEFybzhTZWUyNEhtVkE5d3pldw==

    function CereusTransporter_elasticsearch_send_data($data_array)
    {
        global $settings, $mappings, $options;
        $db_host = read_config_option('cereus_transporter_db_elastic_host');
        $db_port = read_config_option('cereus_transporter_db_elastic_port');
        $db_type   = read_config_option( 'cereus_transporter_dbtype' );
        $db_apikey = read_config_option('cereus_transporter_db_elastic_apikey');
        $db_apiuser = read_config_option('cereus_transporter_db_elastic_user');
        $db_pwd = read_config_option('cereus_transporter_db_elastic_password');



        // Disable error reporting for the DB Connection Creation
        $original_error_reporting = error_reporting();


        // Connect to the remote Elasticsearch host and fetch the database
        try {
            cacti_log("DEBUG: Connecting to ElasticSearch host $db_host at port $db_port.", TRUE, "CereusTransporter");
            if ( strlen($db_apikey) > 0 ) {
                $es_client = ClientBuilder::create()
                                          ->setElasticCloudId($db_apiuser)
                                          ->setApiKey($db_apikey)
                                          ->build();
            } elseif ( strlen($db_pwd) > 0 ) {
                $es_client = ClientBuilder::create()
                                          ->setHosts([$db_host.':'.$db_port])
                                          ->setBasicAuthentication('elastic', $db_pwd)
                                          ->setSSLVerification(false)
                                          ->build();
            } else {
                $es_client = ClientBuilder::create()
                                          ->setHosts([$db_host.':'.$db_port])
                                          ->setSSLVerification(false)
                                          ->build();
            }
        } catch (Exception $e) {
            cacti_log("WARNING: " . $e->getMessage(), TRUE, "CereusTransporter");
            return;
        }

        try {
            $params = array();
            $params['guzzleOptions'] = array(
                \GuzzleHttp\RequestOptions::VERIFY  => false,
                'verify' => false,
                'curl.CURLOPT_SSL_VERIFYPEER' => false
            );
            $transport = $es_client->getTransport();


            // Get ElasticSearch Version
            $response = $es_client->info($params);
            cacti_log("DEBUG: ElasticSearch version: ".  $response->version->number, TRUE, "CereusTransporter");
        } catch (Exception $e) {
            cacti_log("WARNING: " . $e->getMessage(), TRUE, "CereusTransporter");
            return;
        }

        // Connect to the remote Elasticsearch host and fetch the database
        try {
            cacti_log("DEBUG: Creating Index for ElasticSearch.", TRUE, "CereusTransporter");
            $params = [
                'index' => 'cacti_statistics',
                'body' => [
                    'settings' => [
                        'number_of_shards' => 2,
                        'number_of_replicas' => 0
                    ]
                ]
            ];
            $response = $es_client->indices()->create($params);

            if ( $response->getStatusCode() == 200 ) {
                cacti_log("DEBUG: Index 'cacti_statistics' for ElasticSearch has been created.", TRUE, "CereusTransporter");
            } else {
                cacti_log("DEBUG: Failed to create index 'cacti_statistics' for ElasticSearch. Already Exists ?", TRUE, "CereusTransporter");
            }

        } catch (Exception $e) {
            cacti_log("WARNING: " . $e->getMessage(), TRUE, "CereusTransporter");
            //return;
        }

        // Connect to the remote Elasticsearch host and fetch the database
        try {
            cacti_log("DEBUG: Create settings/mapping conenction to ElasticSearch.", TRUE, "CereusTransporter");
            //$index->create(['settings' => $settings, 'mappings' => $mappings], $options);
            //$index->create([],$options);
        } catch (Exception $e) {
            cacti_log("WARNING: " . $e->getMessage(), TRUE, "CereusTransporter");
            return;
        }
        $docs = [];

        cacti_log("DEBUG: Preparing data points.", TRUE, "CereusTransporter");
        // preparing points
        foreach ($data_array as $point) {
            $points = array();
            if (strlen($point['metric']) > 0) {
                if (array_key_exists('metric_text', $point['tags'])) {
                    $point['tags']['metric_text'] = CereusTransporter_cleanTag($db_type, $point['tags']['metric_text']);
                }
                if (array_key_exists('units', $point['tags'])) {
                    $point['tags']['units'] = CereusTransporter_cleanTag($db_type, $point['tags']['units']);
                }
                if (read_config_option('log_verbosity') >= POLLER_VERBOSITY_DEBUG) {
                    cacti_log("DEBUG: Appending the following data to request: [" . $db_type . "] [" . $point['timestamp'] . "] [" . $point['tags']['hostname'] . "] [" . json_encode($point['tags']) . "] [" . $point['tags']['type'] . "] [" . $point['value'] . "]", TRUE, "CereusTransporter");
                }
                $point_str = "";
                try {
                    $docs['body'][] = array(
                        'index' => array(
                            '_index' => 'cacti_statistics'
                        )
                    );
                    $docs['body'][] = array(
                        '@timestamp' => date(DATE_RFC3339, $point['timestamp']), //->format(DateTime::RFC3339),
                        'metric'     => $point['metric'],
                        'value'      => $point['value'],
                        'tags'       => $point['tags'],
                        'fields' => [ 'value' => $point[ 'value' ]]
                    );


                } catch (Exception $e) {
                    cacti_log("ERROR: " . $e->getMessage(), TRUE, "CereusTransporter");
                }
            }

            if (count($docs) > 15000) {
                try {
                    $responses = $es_client->bulk($docs);
                } catch (Exception $e) {
                    cacti_log("WARNING: " . $e->getMessage(), TRUE, "CereusTransporter");
                } finally {
                    unset($responses); // unset  to save memory
                    $docs = [];
                }

            }
        }

        if ($docs) {
            try {
                $responses = $es_client->bulk($docs);
            } catch (Exception $e) {
                cacti_log("WARNING: " . $e->getMessage(), TRUE, "CereusTransporter");
            } finally {
                unset($responses); // unset  to save memory
                $docs = [];
            }
        }

    }

