/*
  Licensed to the Apache Software Foundation (ASF) under one
  or more contributor license agreements.  See the NOTICE file
  distributed with this work for additional information
  regarding copyright ownership.  The ASF licenses this file
  to you under the Apache License, Version 2.0 (the
  "License"); you may not use this file except in compliance
  with the License.  You may obtain a copy of the License at

      http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License.
*/

#include "client_insights.hpp"

#include "config.hpp"
#include "driver_info.hpp"
#include "get_time.hpp"
#include "logger.hpp"
#include "session.hpp"
#include "ssl.hpp"
#include "string.hpp"
#include "utils.hpp"

#include <cassert>
#include <uv.h>

#ifdef _WIN32
#include <windows.h>
#include <winsock2.h>
#else
#include <sys/utsname.h>
#include <unistd.h>
#endif
#define HOSTNAME_MAX_LENGTH 256

#define METADATA_STARTUP_NAME "driver.startup"
#define METADATA_STATUS_NAME "driver.status"
#define METADATA_INSIGHTS_MAPPING_ID "v1"
#define METADATA_LANGUAGE "C/C++"

#define CONFIG_ANTIPATTERN_MSG_MULTI_DC_HOSTS \
  "Contact points contain hosts from "        \
  "multiple data centers but only one "       \
  "is going to be used"
#define CONFIG_ANTIPATTERN_MSG_REMOTE_HOSTS "Using remote hosts for failover"
#define CONFIG_ANTIPATTERN_MSG_DOWNGRADING \
  "Downgrading consistency retry "         \
  "policy in use"
#define CONFIG_ANTIPATTERN_MSG_CERT_VALIDATION \
  "Client-to-node encryption is "              \
  "enabled but server certificate "            \
  "validation is disabled"
#define CONFIG_ANTIPATTERN_MSG_PLAINTEXT_NO_SSL \
  "Plain text authentication is "               \
  "enabled without client-to-node "             \
  "encryption"                                  \
  ""

namespace datastax { namespace internal { namespace core {

MonitorReporting* create_monitor_reporting(const String& client_id, const String& session_id,
                                           const Config& config) {
  // Ensure the client monitor events should be enabled
  unsigned interval_secs = config.monitor_reporting_interval_secs();
  if (interval_secs > 0) {
    return new enterprise::ClientInsights(client_id, session_id, interval_secs);
  }
  return new NopMonitorReporting();
}

}}} // namespace datastax::internal::core

using namespace datastax::internal::core;
using namespace datastax::internal;

namespace datastax { namespace internal { namespace enterprise {

#ifdef _WIN32
#define ERROR_BUFFER_MAX_LENGTH 1024
String get_last_error() {
  DWORD rc = GetLastError();

  char buf[ERROR_BUFFER_MAX_LENGTH];
  size_t size = FormatMessageA(FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS, NULL, rc,
                               MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
                               reinterpret_cast<LPSTR>(&buf[0]), ERROR_BUFFER_MAX_LENGTH, NULL);
  String str(buf, size);
  trim(str);
  return str;
}
#endif

String get_hostname() {
#ifdef _WIN32
  WSADATA data;
  WORD version_required = MAKEWORD(2, 2);
  if (WSAStartup(version_required, &data) != 0) {
    LOG_WARN("Unable to determine hostname: Failed to initialize WinSock2");
    return String();
  }
#endif

  char buf[HOSTNAME_MAX_LENGTH + 1];
  size_t size = HOSTNAME_MAX_LENGTH + 1;
  if (int rc = gethostname(buf, size) != 0) {
    LOG_WARN("Unable to determine hostname: Error code %d", rc);
    return "UNKNOWN";
  }
  return String(buf, size);
}

struct Os {
  String name;
  String version;
  String arch;
};
Os get_os() {
  Os os;
#ifdef _WIN32
  os.name = "Microsoft Windows";

  DWORD size = GetFileVersionInfoSize(TEXT("kernel32.dll"), NULL);
  if (size) {
    Vector<BYTE> version_info(size);
    if (GetFileVersionInfo(TEXT("kernel32.dll"), 0, size, &version_info[0])) {
      VS_FIXEDFILEINFO* file_info = NULL;
      UINT file_info_length = 0;
      if (VerQueryValue(&version_info[0], TEXT("\\"), reinterpret_cast<LPVOID*>(&file_info),
                        &file_info_length)) {
        OStringStream oss;
        oss << static_cast<int>(HIWORD(file_info->dwProductVersionMS)) << "."
            << static_cast<int>(LOWORD(file_info->dwProductVersionMS)) << "."
            << static_cast<int>(HIWORD(file_info->dwProductVersionLS));
        os.version = oss.str();
      } else {
        LOG_DEBUG("Unable to retrieve Windows version: %s\n", get_last_error().c_str());
      }
    } else {
      LOG_DEBUG("Unable to retrieve Windows version (GetFileVersionInfo): %s\n",
                get_last_error().c_str());
    }
  } else {
    LOG_DEBUG("Unable to retrieve Windows version (GetFileVersionInfoSize): %s\n",
              get_last_error().c_str());
  }

#ifdef _WIN64
  os.arch = "x64";
#else
  os.arch = "x86";
#endif
#else
  struct utsname client_info;
  uname(&client_info);
  os.name = client_info.sysname;
  os.version = client_info.release;
  os.arch = client_info.machine;
#endif

  return os;
}

struct Cpus {
  int length;
  String model;
};
Cpus get_cpus() {
  Cpus cpus;

  uv_cpu_info_t* cpus_infos;
  int cpus_count;
  int rc = uv_cpu_info(&cpus_infos, &cpus_count);
  if (rc == 0) {
    uv_cpu_info_t cpus_info = cpus_infos[0];
    cpus.length = cpus_count;
    cpus.model = cpus_info.model;
    uv_free_cpu_info(cpus_infos, cpus_count);
  } else {
    LOG_DEBUG("Unable to determine CPUs information: %s\n", uv_strerror(rc));
  }

  return cpus;
}

class ClientInsightsRequestCallback : public SimpleRequestCallback {
public:
  typedef SharedRefPtr<ClientInsightsRequestCallback> Ptr;

  ClientInsightsRequestCallback(const String& json, const String& event_type)
      : SimpleRequestCallback("CALL InsightsRpc.reportInsight('" + json + "')")
      , event_type_(event_type) {}

  virtual void on_internal_set(ResponseMessage* response) {
    if (response->opcode() != CQL_OPCODE_RESULT) {
      LOG_DEBUG("Failed to send %s event message: Invalid response [%s]", event_type_.c_str(),
                opcode_to_string(response->opcode()).c_str());
    }
  }

  virtual void on_internal_error(CassError code, const String& message) {
    LOG_DEBUG("Failed to send %s event message: %s", event_type_.c_str(), message.c_str());
  }

  virtual void on_internal_timeout() {
    LOG_DEBUG("Failed to send %s event message: Timed out waiting for response",
              event_type_.c_str());
  }

private:
  String event_type_;
};

void metadata(ClientInsights::Writer& writer, const String& name) {
  writer.Key("metadata");
  writer.StartObject();

  writer.Key("name");
  writer.String(name.c_str());
  writer.Key("insightMappingId");
  writer.String(METADATA_INSIGHTS_MAPPING_ID);
  writer.Key("insightType");
  writer.String("EVENT"); // TODO: Make this an enumeration in the future
  writer.Key("timestamp");
  writer.Uint64(get_time_since_epoch_ms());
  writer.Key("tags");
  writer.StartObject();
  writer.Key("language");
  writer.String(METADATA_LANGUAGE);
  writer.EndObject(); // tags

  writer.EndObject();
}

class StartupMessageHandler : public RefCounted<StartupMessageHandler> {
public:
  typedef SharedRefPtr<StartupMessageHandler> Ptr;

  StartupMessageHandler(const Connection::Ptr& connection, const String& client_id,
                        const String& session_id, const Config& config, const HostMap& hosts,
                        const LoadBalancingPolicy::Vec& initialized_policies)
      : connection_(connection)
      , client_id_(client_id)
      , session_id_(session_id)
      , config_(config)
      , hosts_(hosts)
      , initialized_policies_(initialized_policies) {}

  ~StartupMessageHandler() {
    ClientInsights::StringBuffer buffer;
    ClientInsights::Writer writer(buffer);

    writer.StartObject();
    metadata(writer, METADATA_STARTUP_NAME);
    startup_message_data(writer);
    writer.EndObject();

    assert(writer.IsComplete() && "Startup JSON is incomplete");
    connection_->write_and_flush(RequestCallback::Ptr(
        new ClientInsightsRequestCallback(buffer.GetString(), METADATA_STARTUP_NAME)));
  }

  void send_message() { resolve_contact_points(); }

private:
  // Startup message associated methods
  void startup_message_data(ClientInsights::Writer& writer) {
    writer.Key("data");
    writer.StartObject();

    writer.Key("clientId");
    writer.String(client_id_.c_str());
    writer.Key("sessionId");
    writer.String(session_id_.c_str());
    bool is_application_name_generated = false;
    writer.Key("applicationName");
    if (!config_.application_name().empty()) {
      writer.String(config_.application_name().c_str());
    } else {
      writer.String(driver_name());
      is_application_name_generated = true;
    }
    writer.Key("applicationNameWasGenerated");
    writer.Bool(is_application_name_generated);
    if (!config_.application_version().empty()) {
      writer.Key("applicationVersion");
      writer.String(config_.application_version().c_str());
    }
    writer.Key("driverName");
    writer.String(driver_name());
    writer.Key("driverVersion");
    writer.String(driver_version());
    contact_points(writer);
    data_centers(writer);
    writer.Key("initialControlConnection");
    writer.String(connection_->resolved_address().to_string(true).c_str());
    writer.Key("protocolVersion");
    writer.Int(connection_->protocol_version().value());
    writer.Key("localAddress");
    writer.String(get_local_address(connection_->handle()).c_str());
    writer.Key("hostName");
    writer.String(get_hostname().c_str());
    execution_profiles(writer);
    pool_size_by_host_distance(writer);
    writer.Key("heartbeatInterval");
    writer.Uint64(config_.connection_heartbeat_interval_secs() * 1000); // in milliseconds
    writer.Key("compression");
    writer.String("NONE"); // TODO: Update once compression is added
    reconnection_policy(writer);
    ssl(writer);
    auth_provider(writer);
    other_options(writer);
    platform_info(writer);
    config_anti_patterns(writer);
    writer.Key("periodicStatusInterval");
    writer.Uint(config_.monitor_reporting_interval_secs());

    writer.EndObject();
  }

  void contact_points(ClientInsights::Writer& writer) {
    writer.Key("contactPoints");
    writer.StartObject();

    for (ResolvedHostMap::const_iterator map_it = contact_points_resolved_.begin(),
                                         map_end = contact_points_resolved_.end();
         map_it != map_end; ++map_it) {
      writer.Key(map_it->first.c_str());
      writer.StartArray();
      for (AddressSet::const_iterator vec_it = map_it->second.begin(),
                                      vec_end = map_it->second.end();
           vec_it != vec_end; ++vec_it) {
        writer.String(vec_it->to_string(true).c_str());
      }
      writer.EndArray();
    }

    writer.EndObject();
  }

  void data_centers(ClientInsights::Writer& writer) {
    writer.Key("dataCenters");
    writer.StartArray();

    Set<String> data_centers;
    for (HostMap::const_iterator it = hosts_.begin(), end = hosts_.end(); it != end; ++it) {
      const String& data_center = it->second->dc();
      if (data_centers.insert(data_center).second) {
        writer.String(data_center.c_str());
      }
    }

    writer.EndArray();
  }

  void execution_profiles(ClientInsights::Writer& writer) {
    writer.Key("executionProfiles");
    writer.StartObject();

    const ExecutionProfile& default_profile = config_.default_profile();
    const ExecutionProfile::Map& profiles = config_.profiles();
    writer.Key("default");
    execution_profile_as_json(writer, default_profile);
    for (ExecutionProfile::Map::const_iterator it = profiles.begin(), end = profiles.end();
         it != end; ++it) {
      writer.Key(it->first.c_str());
      execution_profile_as_json(writer, it->second, &default_profile);
    }

    writer.EndObject();
  }

  void pool_size_by_host_distance(ClientInsights::Writer& writer) {
    writer.Key("poolSizeByHostDistance");
    writer.StartObject();

    writer.Key("local");
    writer.Uint(config_.core_connections_per_host() * hosts_.size());
    // NOTE: Remote does not pertain to DataStax C/C++ driver pool
    writer.Key("remote");
    writer.Uint(0);

    writer.EndObject();
  }

  void reconnection_policy(ClientInsights::Writer& writer) {
    writer.Key("reconnectionPolicy");
    writer.StartObject();

    ReconnectionPolicy::Ptr reconnection_policy = config_.reconnection_policy();
    writer.Key("type");
    if (reconnection_policy->type() == ReconnectionPolicy::CONSTANT) {
      writer.String("ConstantReconnectionPolicy");
    } else if (reconnection_policy->type() == ReconnectionPolicy::EXPONENTIAL) {
      writer.String("ExponentialReconnectionPolicy");
    } else {
      assert(false && "Reconnection policy needs to be added");
      writer.String("UnknownReconnectionPolicy");
    }
    writer.Key("options");
    writer.StartObject();
    if (reconnection_policy->type() == ReconnectionPolicy::CONSTANT) {
      ConstantReconnectionPolicy::Ptr crp =
          static_cast<ConstantReconnectionPolicy::Ptr>(reconnection_policy);
      writer.Key("delayMs");
      writer.Uint(crp->delay_ms());
    } else if (reconnection_policy->type() == ReconnectionPolicy::EXPONENTIAL) {
      ExponentialReconnectionPolicy::Ptr erp =
          static_cast<ExponentialReconnectionPolicy::Ptr>(reconnection_policy);
      writer.Key("baseDelayMs");
      writer.Uint(erp->base_delay_ms());
      writer.Key("maxDelayMs");
      writer.Uint(erp->max_delay_ms());
    }
    writer.EndObject(); // options

    writer.EndObject();
  }

  void ssl(ClientInsights::Writer& writer) {
    writer.Key("ssl");
    writer.StartObject();

    const SslContext::Ptr& ssl_context = config_.ssl_context();
    writer.Key("enabled");
    if (ssl_context) {
      writer.Bool(true);
    } else {
      writer.Bool(false);
    }
    writer.Key("certValidation");
    if (ssl_context) {
      writer.Bool(ssl_context->is_cert_validation_enabled());
    } else {
      writer.Bool(false);
    }

    writer.EndObject();
  }

  void auth_provider(ClientInsights::Writer& writer) {
    const AuthProvider::Ptr& auth_provider = config_.auth_provider();
    if (auth_provider) {
      writer.Key("authProvider");
      writer.StartObject();

      writer.Key("type");
      writer.String(auth_provider->name().c_str());

      writer.EndObject();
    }
  }

  void other_options(ClientInsights::Writer& writer) {
    writer.Key("otherOptions");
    writer.StartObject();

    writer.Key("configuration");
    writer.StartObject();
    writer.Key("protocolVersion");
    writer.Uint(config_.protocol_version().value());
    writer.Key("useBetaProtocol");
    writer.Bool(config_.use_beta_protocol_version());
    writer.Key("threadCountIo");
    writer.Uint(config_.thread_count_io());
    writer.Key("queueSizeIo");
    writer.Uint(config_.queue_size_io());
    writer.Key("coreConnectionsPerHost");
    writer.Uint(config_.core_connections_per_host());
    writer.Key("connectTimeoutMs");
    writer.Uint(config_.connect_timeout_ms());
    writer.Key("resolveTimeoutMs");
    writer.Uint(config_.resolve_timeout_ms());
    writer.Key("maxSchemaWaitTimeMs");
    writer.Uint(config_.max_schema_wait_time_ms());
    writer.Key("maxTracingWaitTimeMs");
    writer.Uint(config_.max_tracing_wait_time_ms());
    writer.Key("tracingConsistency");
    writer.String(cass_consistency_string(config_.tracing_consistency()));
    writer.Key("coalesceDelayUs");
    writer.Uint64(config_.coalesce_delay_us());
    writer.Key("newRequestRatio");
    writer.Uint(config_.new_request_ratio());
    writer.Key("logLevel");
    writer.String(cass_log_level_string(config_.log_level()));
    writer.Key("tcpNodelayEnable");
    writer.Bool(config_.tcp_nodelay_enable());
    writer.Key("tcpKeepaliveEnable");
    writer.Bool(config_.tcp_keepalive_enable());
    writer.Key("tcpKeepaliveDelaySecs");
    writer.Uint(config_.tcp_keepalive_delay_secs());
    writer.Key("connectionIdleTimeoutSecs");
    writer.Uint(config_.connection_idle_timeout_secs());
    writer.Key("useSchema");
    writer.Bool(config_.use_schema());
    writer.Key("useHostnameResolution");
    writer.Bool(config_.use_hostname_resolution());
    writer.Key("useRandomizedContactPoints");
    writer.Bool(config_.use_randomized_contact_points());
    writer.Key("maxReusableWriteObjects");
    writer.Uint(config_.max_reusable_write_objects());
    writer.Key("prepareOnAllHosts");
    writer.Bool(config_.prepare_on_all_hosts());
    writer.Key("prepareOnUpOrAddHost");
    writer.Bool(config_.prepare_on_up_or_add_host());
    writer.Key("noCompact");
    writer.Bool(config_.no_compact());
    writer.Key("cloudSecureConnectBundleLoaded");
    writer.Bool(config_.cloud_secure_connection_config().is_loaded());
    writer.Key("clusterMetadataResolver");
    writer.String(config_.cluster_metadata_resolver_factory()->name());
    writer.EndObject(); // configuration

    writer.EndObject(); // otherOptions
  }

  void platform_info(ClientInsights::Writer& writer) {
    writer.Key("platformInfo");
    writer.StartObject();

    writer.Key("os");
    writer.StartObject();
    Os os = get_os();
    writer.Key("name");
    writer.String(os.name.c_str());
    writer.Key("version");
    writer.String(os.version.c_str());
    writer.Key("arch");
    writer.String(os.arch.c_str());
    writer.EndObject(); // os

    writer.Key("cpus");
    writer.StartObject();
    Cpus cpus = get_cpus();
    writer.Key("length");
    writer.Int(cpus.length);
    writer.Key("model");
    writer.String(cpus.model.c_str());
    writer.EndObject(); // cpus

    writer.Key("runtime");
    writer.StartObject();
#if defined(__clang__) || defined(__APPLE_CC__)
    writer.Key("Clang/LLVM");
    writer.String(STRINGIFY(__clang_major__) "." STRINGIFY(__clang_minor__) "." STRINGIFY(
        __clang_patchlevel__));
#elif defined(__INTEL_COMPILER)
    writer.Key("Intel ICC/ICPC");
    writer.String(STRINGIFY(__INTEL_COMPILER));
#elif defined(__GNUC__) || defined(__GNUG__)
    writer.Key("GNU GCC/G++");
    writer.String(
        STRINGIFY(__GNUC__) "." STRINGIFY(__GNUC_MINOR__) "." STRINGIFY(__GNUC_PATCHLEVEL__));
#elif defined(__HP_aCC)
    writer.Key("Hewlett-Packard C/aC++");
    writer.String(STRINGIFY(__HP_aCC));
#elif defined(__IBMCPP__)
    writer.Key("IBM XL C/C++");
    writer.String(STRINGIFY(__xlc__));
#elif defined(_MSC_VER)
    writer.Key("Microsoft Visual Studio");
    writer.String(STRINGIFY(_MSC_FULL_VER));
#elif defined(__PGI)
    writer.Key("Portland Group PGCC/PGCPP");
    writer.String(
        STRINGIFY(__PGIC__) "." STRINGIFY(__PGIC_MINOR__) "." STRINGIFY(__PGIC_PATCHLEVEL__));
#elif defined(__SUNPRO_CC)
    writer.Key("Oracle Solaris Studio");
    writer.String(STRINGIFY(__SUNPRO_CC));
#else
    writer.Key("Unknown");
    writer.String("Unknown");
#endif
    writer.Key("uv");
    writer.String(STRINGIFY(UV_VERSION_MAJOR) "." STRINGIFY(UV_VERSION_MINOR) "." STRINGIFY(
        UV_VERSION_PATCH));
    writer.Key("openssl");
#ifdef OPENSSL_VERSION_TEXT
    writer.String(OPENSSL_VERSION_TEXT);
#else
#ifdef LIBRESSL_VERSION_NUMBER
    writer.String("LibreSSL " STRINGIFY(LIBRESSL_VERSION_NUMBER));
#else
    writer.String("OpenSSL " STRINGIFY(OPENSSL_VERSION_NUMBER));
#endif
#endif
    writer.EndObject(); // runtime

    writer.EndObject(); // platformInfo
  }

  void config_anti_patterns(ClientInsights::Writer& writer) {
    StringPairVec config_anti_patterns = get_config_anti_patterns(
        config_.default_profile(), config_.profiles(), initialized_policies_, hosts_,
        config_.ssl_context(), config_.auth_provider());
    if (!config_anti_patterns.empty()) {
      writer.Key("configAntiPatterns");
      writer.StartObject();

      for (StringPairVec::const_iterator it = config_anti_patterns.begin(),
                                         end = config_anti_patterns.end();
           it != end; ++it) {
        writer.Key(it->first.c_str());
        writer.String(it->second.c_str());
      }

      writer.EndObject();
    }
  }

private:
  // Startup message helper methods
  void resolve_contact_points() {
    const AddressVec& contact_points = config_.contact_points();
    const int port = config_.port();
    MultiResolver::Ptr resolver;

    for (AddressVec::const_iterator it = contact_points.begin(), end = contact_points.end();
         it != end; ++it) {
      const Address& contact_point = *it;
      // Attempt to parse the contact point string. If it's an IP address
      // then immediately add it to our resolved contact points, otherwise
      // attempt to resolve the string as a hostname.
      if (contact_point.is_resolved()) {
        AddressSet addresses;
        addresses.insert(contact_point);
        contact_points_resolved_[contact_point.hostname_or_address()] = addresses;
      } else {
        if (!resolver) {
          inc_ref();
          resolver.reset(
              new MultiResolver(bind_callback(&StartupMessageHandler::on_resolve, this)));
        }
        resolver->resolve(connection_->loop(), contact_point.hostname_or_address(), port,
                          config_.resolve_timeout_ms());
      }
    }

    // NOTE: If no resolution is performed the startup message will be sent in
    //       the destructor
  }

  void on_resolve(MultiResolver* resolver) {
    const Resolver::Vec& resolvers = resolver->resolvers();
    for (Resolver::Vec::const_iterator it = resolvers.begin(), end = resolvers.end(); it != end;
         ++it) {
      const Resolver::Ptr resolver(*it);
      AddressSet addresses;
      if (resolver->is_success()) {
        if (!resolver->addresses().empty()) {
          for (AddressVec::const_iterator it = resolver->addresses().begin(),
                                          end = resolver->addresses().end();
               it != end; ++it) {
            addresses.insert(*it);
          }
        }
      }
      contact_points_resolved_[resolver->hostname()] = addresses; // Empty resolved addresses are OK
    }

    dec_ref(); // Send startup message in destructor
  }

  String get_local_address(const uv_tcp_t* tcp) const {
    Address::SocketStorage name;
    int namelen = sizeof(name);
    if (uv_tcp_getsockname(tcp, name.addr(), &namelen) == 0) {
      Address address(name.addr());
      if (address.is_valid_and_resolved()) {
        return address.to_string();
      }
    }
    return "unknown";
  }

  void execution_profile_as_json(ClientInsights::Writer& writer, const ExecutionProfile& profile,
                                 const ExecutionProfile* default_profile = NULL) {
    writer.StartObject();

    if (!default_profile || (default_profile && profile.request_timeout_ms() !=
                                                    default_profile->request_timeout_ms())) {
      writer.Key("requestTimeoutMs");
      writer.Uint64(profile.request_timeout_ms());
    }
    if (!default_profile ||
        (default_profile && profile.consistency() != default_profile->consistency())) {
      writer.Key("consistency");
      writer.String(cass_consistency_string(profile.consistency()));
    }
    if (!default_profile || (default_profile && profile.serial_consistency() !=
                                                    default_profile->serial_consistency())) {
      writer.Key("serialConsistency");
      writer.String(cass_consistency_string(profile.serial_consistency()));
    }
    if (!default_profile ||
        (default_profile && profile.retry_policy() != default_profile->retry_policy())) {
      const RetryPolicy::Ptr& retry_policy = profile.retry_policy();
      if (retry_policy) {
        writer.Key("retryPolicy");
        if (retry_policy->type() == RetryPolicy::DEFAULT) {
          writer.String("DefaultRetryPolicy");
        } else if (retry_policy->type() == RetryPolicy::DOWNGRADING) {
          writer.String("DowngradingConsistencyRetryPolicy");
        } else if (retry_policy->type() == RetryPolicy::FALLTHROUGH) {
          writer.String("FallthroughRetryPolicy");
        } else if (retry_policy->type() == RetryPolicy::LOGGING) {
          writer.String("LoggingRetryPolicy");
        } else {
          LOG_DEBUG("Invalid retry policy: %d", retry_policy->type());
          writer.String("unknown");
        }
      }
    }

    if (profile.load_balancing_policy()) {
      writer.Key("loadBalancing");
      writer.StartObject();
      writer.Key("type");
      LoadBalancingPolicy* current_lbp = profile.load_balancing_policy().get();
      do {
        // NOTE: DCAware and RoundRobin are leaf policies (e.g. not chainable)
        if (dynamic_cast<DCAwarePolicy*>(current_lbp)) {
          writer.String("DCAwarePolicy");
          break;
        } else if (dynamic_cast<RoundRobinPolicy*>(current_lbp)) {
          writer.String("RoundRobinPolicy");
          break;
        }

        if (ChainedLoadBalancingPolicy* chained_lbp =
                dynamic_cast<ChainedLoadBalancingPolicy*>(current_lbp)) {
          current_lbp = chained_lbp->child_policy().get();
        } else {
          current_lbp = NULL;
        }
      } while (current_lbp);
      writer.Key("options");
      writer.StartObject();
      if (DCAwarePolicy* dc_lbp = dynamic_cast<DCAwarePolicy*>(current_lbp)) {
        writer.Key("localDc");
        if (dc_lbp->local_dc().empty()) {
          writer.Null();
        } else {
          writer.String(dc_lbp->local_dc().c_str());
        }
        writer.Key("usedHostsPerRemoteDc");
        writer.Uint64(dc_lbp->used_hosts_per_remote_dc());
        writer.Key("allowRemoteDcsForLocalCl");
        writer.Bool(!dc_lbp->skip_remote_dcs_for_local_cl());
      }
      if (!profile.blacklist().empty()) {
        writer.Key("blacklist");
        writer.String(implode(profile.blacklist()).c_str());
      }
      if (!profile.blacklist_dc().empty()) {
        writer.Key("blacklistDc");
        writer.String(implode(profile.blacklist_dc()).c_str());
      }
      if (!profile.whitelist().empty()) {
        writer.Key("whitelist");
        writer.String(implode(profile.whitelist()).c_str());
      }
      if (!profile.whitelist_dc().empty()) {
        writer.Key("whitelistDc");
        writer.String(implode(profile.whitelist_dc()).c_str());
      }
      if (profile.token_aware_routing()) {
        writer.Key("tokenAwareRouting");
        writer.StartObject();
        writer.Key("shuffleReplicas");
        writer.Bool(profile.token_aware_routing_shuffle_replicas());
        writer.EndObject(); // tokenAwareRouting
      }
      if (profile.latency_aware()) {
        writer.Key("latencyAwareRouting");
        writer.StartObject();
        writer.Key("exclusionThreshold");
        writer.Double(profile.latency_aware_routing_settings().exclusion_threshold);
        writer.Key("scaleNs");
        writer.Uint64(profile.latency_aware_routing_settings().scale_ns);
        writer.Key("retryPeriodNs");
        writer.Uint64(profile.latency_aware_routing_settings().retry_period_ns);
        writer.Key("updateRateMs");
        writer.Uint64(profile.latency_aware_routing_settings().update_rate_ms);
        writer.Key("minMeasured");
        writer.Uint64(profile.latency_aware_routing_settings().min_measured);
        writer.EndObject(); // latencyAwareRouting
      }
      writer.EndObject(); // options

      writer.EndObject(); // loadBalancingPolicy
    }

    typedef ConstantSpeculativeExecutionPolicy CSEP;
    CSEP* default_csep =
        default_profile ? dynamic_cast<CSEP*>(default_profile->speculative_execution_policy().get())
                        : NULL;
    CSEP* csep = dynamic_cast<CSEP*>(profile.speculative_execution_policy().get());
    if (csep) {
      if (!default_csep ||
          (default_csep->constant_delay_ms_ != csep->constant_delay_ms_ &&
           default_csep->max_speculative_executions_ != csep->max_speculative_executions_)) {
        writer.Key("speculativeExecutionPolicy");
        writer.StartObject();
        writer.Key("type");
        writer.String("ConstantSpeculativeExecutionPolicy");

        writer.Key("options");
        writer.StartObject();
        writer.Key("constantDelayMs");
        writer.Uint64(csep->constant_delay_ms_);
        writer.Key("maxSpeculativeExecutions");
        writer.Int(csep->max_speculative_executions_);
        writer.EndObject(); // options

        writer.EndObject(); // speculativeExecutionPolicy
      }
    }

    writer.EndObject(); // executionProfile
  }

  typedef std::pair<String, String> StringPair;
  typedef Vector<StringPair> StringPairVec;
  StringPairVec get_config_anti_patterns(const ExecutionProfile& default_profile,
                                         const ExecutionProfile::Map& profiles,
                                         const LoadBalancingPolicy::Vec& policies,
                                         const HostMap& hosts, const SslContext::Ptr& ssl_context,
                                         const AuthProvider::Ptr& auth_provider) {
    StringPairVec config_anti_patterns;

    if (is_contact_points_multiple_dcs(policies, hosts)) {
      config_anti_patterns.push_back(
          StringPair("contactPointsMultipleDCs", CONFIG_ANTIPATTERN_MSG_MULTI_DC_HOSTS));
      LOG_WARN("Configuration anti-pattern detected: %s", CONFIG_ANTIPATTERN_MSG_MULTI_DC_HOSTS);
    }

    for (LoadBalancingPolicy::Vec::const_iterator it = policies.begin(), end = policies.end();
         it != end; ++it) {
      DCAwarePolicy* dc_lbp = get_dc_aware_policy(*it);
      if (dc_lbp && !dc_lbp->skip_remote_dcs_for_local_cl()) {
        config_anti_patterns.push_back(
            StringPair("useRemoteHosts", CONFIG_ANTIPATTERN_MSG_REMOTE_HOSTS));
        LOG_WARN("Configuration anti-pattern detected: %s", CONFIG_ANTIPATTERN_MSG_REMOTE_HOSTS);
        break;
      }
    }

    bool is_downgrading_consistency_enabled =
        is_downgrading_retry_anti_pattern(default_profile.retry_policy());
    if (!is_downgrading_consistency_enabled) {
      for (ExecutionProfile::Map::const_iterator it = profiles.begin(), end = profiles.end();
           it != end; ++it) {
        if (is_downgrading_retry_anti_pattern(it->second.retry_policy())) {
          is_downgrading_consistency_enabled = true;
          break;
        }
      }
    }
    if (is_downgrading_consistency_enabled) {
      config_anti_patterns.push_back(
          StringPair("downgradingConsistency", CONFIG_ANTIPATTERN_MSG_DOWNGRADING));
      LOG_WARN("Configuration anti-pattern detected: %s", CONFIG_ANTIPATTERN_MSG_DOWNGRADING);
    }

    if (ssl_context && !ssl_context->is_cert_validation_enabled()) {
      config_anti_patterns.push_back(
          StringPair("sslWithoutCertValidation", CONFIG_ANTIPATTERN_MSG_CERT_VALIDATION));
      LOG_WARN("Configuration anti-pattern detected: %s", CONFIG_ANTIPATTERN_MSG_CERT_VALIDATION);
    }

    if (auth_provider && auth_provider->name().find("PlainTextAuthProvider") != String::npos &&
        !ssl_context) {
      config_anti_patterns.push_back(
          StringPair("plainTextAuthWithoutSsl", CONFIG_ANTIPATTERN_MSG_PLAINTEXT_NO_SSL));
      LOG_WARN("Configuration anti-pattern detected: %s", CONFIG_ANTIPATTERN_MSG_PLAINTEXT_NO_SSL);
    }

    return config_anti_patterns;
  }

  DCAwarePolicy* get_dc_aware_policy(const LoadBalancingPolicy::Ptr& policy) {
    LoadBalancingPolicy* current_lbp = policy.get();
    do {
      if (DCAwarePolicy* dc_lbp = dynamic_cast<DCAwarePolicy*>(current_lbp)) {
        return dc_lbp;
      }
      if (ChainedLoadBalancingPolicy* chained_lbp =
              dynamic_cast<ChainedLoadBalancingPolicy*>(current_lbp)) {
        current_lbp = chained_lbp->child_policy().get();
      } else {
        break;
      }
    } while (current_lbp);

    return NULL;
  }

  bool is_contact_points_multiple_dcs(const LoadBalancingPolicy::Vec& policies,
                                      const HostMap& hosts) {
    // Get the DC aware load balancing policy if it is the only policy that
    // exists. If found this policy will be used after the contact points have
    // been resolved in order to determine if there are contacts points that exist
    // in multiple DCs using a copy of the discovered hosts.
    if (policies.size() == 1) {
      DCAwarePolicy* policy = get_dc_aware_policy(policies[0]);
      if (policy) {
        // Loop through the resolved contacts, find the correct initialized host
        // and if the contact point is a remote host identify as an anti-pattern
        for (ResolvedHostMap::const_iterator resolved_it = contact_points_resolved_.begin(),
                                             hosts_end = contact_points_resolved_.end();
             resolved_it != hosts_end; ++resolved_it) {
          const AddressSet& addresses = resolved_it->second;
          for (AddressSet::const_iterator addresses_it = addresses.begin(),
                                          addresses_end = addresses.end();
               addresses_it != addresses_end; ++addresses_it) {
            const Address& address = *addresses_it;
            for (HostMap::const_iterator hosts_it = hosts.begin(), hosts_end = hosts.end();
                 hosts_it != hosts_end; ++hosts_it) {
              const Host::Ptr& host = hosts_it->second;
              if (host->address() == address &&
                  policy->distance(host) == CASS_HOST_DISTANCE_REMOTE) {
                return true;
              }
            }
          }
        }
      }
    }

    return false;
  }

  bool is_downgrading_retry_anti_pattern(const RetryPolicy::Ptr& policy) {
    if (policy && policy->type() == RetryPolicy::DOWNGRADING) {
      return true;
    }
    return false;
  }

private:
  const Connection::Ptr connection_;
  const String client_id_;
  const String session_id_;
  const Config config_;
  const HostMap hosts_;
  const LoadBalancingPolicy::Vec initialized_policies_;

private:
  typedef Map<String, AddressSet> ResolvedHostMap;
  ResolvedHostMap contact_points_resolved_;
};

ClientInsights::ClientInsights(const String& client_id, const String& session_id,
                               unsigned interval_secs)
    : client_id_(client_id)
    , session_id_(session_id)
    , interval_ms_(interval_secs * 1000) {}

uint64_t ClientInsights::interval_ms(const VersionNumber& dse_server_version) const {
  // DSE v5.1.13+ (backported)
  // DSE v6.0.5+ (backported)
  // DSE v6.7.0 was the first to supported the Insights RPC call
  if ((dse_server_version >= VersionNumber(5, 1, 13) &&
       dse_server_version < VersionNumber(6, 0, 0)) ||
      dse_server_version >= VersionNumber(6, 0, 5)) {
    return interval_ms_;
  }
  return 0;
}

void ClientInsights::send_startup_message(const Connection::Ptr& connection, const Config& config,
                                          const HostMap& hosts,
                                          const LoadBalancingPolicy::Vec& initialized_policies) {
  StartupMessageHandler::Ptr handler = StartupMessageHandler::Ptr(new StartupMessageHandler(
      connection, client_id_, session_id_, config, hosts, initialized_policies));
  handler->send_message();
}

void ClientInsights::send_status_message(const Connection::Ptr& connection, const HostMap& hosts) {
  StringBuffer buffer;
  Writer writer(buffer);

  writer.StartObject();
  metadata(writer, METADATA_STATUS_NAME);

  writer.Key("data");
  writer.StartObject();

  writer.Key("clientId");
  writer.String(client_id_.c_str());
  writer.Key("sessionId");
  writer.String(session_id_.c_str());
  writer.Key("controlConnection");
  writer.String(connection->resolved_address().to_string(true).c_str());

  writer.Key("conntectedNodes");
  writer.StartObject();
  for (HostMap::const_iterator it = hosts.begin(), end = hosts.end(); it != end; ++it) {
    String address_with_port = it->first.to_string(true);
    const Host::Ptr& host = it->second;
    writer.Key(address_with_port.c_str());
    writer.StartObject();
    writer.Key("connections");
    writer.Int(host->connection_count());
    writer.Key("inFlightQueries");
    writer.Int(host->inflight_request_count());
    writer.EndObject(); // address_with_port
  }
  writer.EndObject(); // connectedNodes

  writer.EndObject(); // data
  writer.EndObject();

  assert(writer.IsComplete() && "Status JSON is incomplete");
  connection->write_and_flush(RequestCallback::Ptr(
      new ClientInsightsRequestCallback(buffer.GetString(), METADATA_STATUS_NAME)));
}

}}} // namespace datastax::internal::enterprise