using System;
using System.Diagnostics;
using System.Reactive;
using System.Reactive.Subjects;
using System.Text;
using System.Threading.Tasks;
using InfluxDB.Client.Api.Client;
using InfluxDB.Client.Api.Domain;
using InfluxDB.Client.Api.Service;
using InfluxDB.Client.Core;
using InfluxDB.Client.Core.Exceptions;
using InfluxDB.Client.Core.Internal;
using InfluxDB.Client.Internal;
namespace InfluxDB.Client
{
public interface IInfluxDBClient : IDisposable
{
///
/// Get the Query client.
///
/// the mapper used for mapping FluxResults to POCO
/// the new client instance for the Query API
IQueryApi GetQueryApi(IDomainObjectMapper mapper = null);
///
/// Get the synchronous version of Query client.
///
/// the mapper used for mapping FluxResults to POCO
/// the new synchronous client instance for the Query API
IQueryApiSync GetQueryApiSync(IDomainObjectMapper mapper = null);
///
/// Get the Write client.
///
/// the mapper used for mapping to PointData
/// the new client instance for the Write API
IWriteApi GetWriteApi(IDomainObjectMapper mapper = null);
///
/// Get the Write client.
///
/// the configuration for a write client
/// the converter used for mapping to PointData
/// the new client instance for the Write API
IWriteApi GetWriteApi(WriteOptions writeOptions, IDomainObjectMapper mapper = null);
///
/// Get the Write async client.
///
/// the converter used for mapping to PointData
/// the new client instance for the Write API Async without batching
IWriteApiAsync GetWriteApiAsync(IDomainObjectMapper mapper = null);
///
/// Get the client.
///
/// the new client instance for Organization API
IOrganizationsApi GetOrganizationsApi();
///
/// Get the client.
///
/// the new client instance for User API
IUsersApi GetUsersApi();
///
/// Get the client.
///
/// the new client instance for Bucket API
IBucketsApi GetBucketsApi();
///
/// Get the client.
///
/// the new client instance for Source API
ISourcesApi GetSourcesApi();
///
/// Get the client.
///
/// the new client instance for Authorization API
IAuthorizationsApi GetAuthorizationsApi();
///
/// Get the client.
///
/// the new client instance for Task API
ITasksApi GetTasksApi();
///
/// Get the client.
///
/// the new client instance for Scraper API
IScraperTargetsApi GetScraperTargetsApi();
///
/// Get the client.
///
/// the new client instance for Telegrafs API
ITelegrafsApi GetTelegrafsApi();
///
/// Get the client.
///
/// the new client instance for Label API
ILabelsApi GetLabelsApi();
///
/// Get the client.
///
/// the new client instance for NotificationEndpoint API
INotificationEndpointsApi GetNotificationEndpointsApi();
///
/// Get the client.
///
/// the new client instance for NotificationRules API
INotificationRulesApi GetNotificationRulesApi();
///
/// Get the client.
///
/// the new client instance for Checks API
IChecksApi GetChecksApi();
///
/// Get the Delete client.
///
/// the new client instance for Delete API
IDeleteApi GetDeleteApi();
///
/// Create an InvokableScripts API instance.
///
/// the mapper used for mapping invocation results to POCO
/// New instance of InvokableScriptsApi.
IInvokableScriptsApi GetInvokableScriptsApi(IDomainObjectMapper mapper = null);
///
/// Create a service for specified type.
///
/// type of service
/// type of service
/// new instance of service
TS CreateService(Type serviceType) where TS : IApiAccessor;
///
/// Set the log level for the request and response information.
///
/// the log level to set
void SetLogLevel(LogLevel logLevel);
///
/// Set the that is used for logging requests and responses.
///
/// Log Level
LogLevel GetLogLevel();
///
/// Enable Gzip compress for http requests.
///
/// Currently only the "Write" and "Query" endpoints supports the Gzip compression.
///
///
IInfluxDBClient EnableGzip();
///
/// Disable Gzip compress for http request body.
///
/// this
IInfluxDBClient DisableGzip();
///
/// Returns whether Gzip compress for http request body is enabled.
///
/// true if gzip is enabled.
bool IsGzipEnabled();
///
/// Get the health of an instance.
///
/// health of an instance
[Obsolete("This method is obsolete. Call 'PingAsync()' instead.", false)]
Task HealthAsync();
///
/// Check the status of InfluxDB Server.
///
/// true if server is healthy otherwise return false
Task PingAsync();
///
/// Return the version of the connected InfluxDB Server.
///
/// the version String, otherwise unknown
/// throws when request did not succesfully ends
Task VersionAsync();
///
/// Check the readiness of InfluxDB Server at startup. It is not supported by InfluxDB Cloud.
///
/// return null if the InfluxDB is not ready
Task ReadyAsync();
///
/// Post onboarding request, to setup initial user, org and bucket.
///
/// to setup defaults
/// With status code 422 when an onboarding has already been completed
/// defaults for first run
Task OnboardingAsync(OnboardingRequest onboarding);
///
/// Check if database has default user, org, bucket created, returns true if not.
///
/// True if onboarding has already been completed otherwise false
Task IsOnboardingAllowedAsync();
}
public class InfluxDBClient : AbstractRestClient, IInfluxDBClient
{
private readonly ApiClient _apiClient;
private readonly ExceptionFactory _exceptionFactory;
private readonly HealthService _healthService;
private readonly LoggingHandler _loggingHandler;
private readonly GzipHandler _gzipHandler;
private readonly ReadyService _readyService;
private readonly PingService _pingService;
private readonly SetupService _setupService;
private readonly InfluxDBClientOptions _options;
private readonly Subject _disposeNotification = new Subject();
///
/// Create a instance of the InfluxDB 2.x client. The url could be a connection string with various configurations.
///
/// e.g.: "http://localhost:8086?timeout=5000&logLevel=BASIC
/// The following options are supported:
///
/// - org - default destination organization for writes and queries
/// - bucket - default destination bucket for writes
/// - token - the token to use for the authorization
/// - logLevel (default - NONE) - rest client verbosity level
/// - timeout (default - 10000) - The timespan to wait before the HTTP request times out in milliseconds
/// - allowHttpRedirects (default - false) - Configure automatically following HTTP 3xx redirects
/// - verifySsl (default - true) - Ignore Certificate Validation Errors when false
///
/// Options for logLevel:
///
/// - Basic - Logs request and response lines.
/// - Body - Logs request and response lines including headers and body (if present). Note that applying the `Body` LogLevel will disable chunking while streaming and will load the whole response into memory.
/// - Headers - Logs request and response lines including headers.
/// - None - Disable logging.
///
///
///
/// connection string with various configurations
public InfluxDBClient(string url) :
this(new InfluxDBClientOptions(url))
{
}
///
/// Create a instance of the InfluxDB 2.x client.
///
/// the url to connect to the InfluxDB 2.x
/// the username to use in the basic auth
/// the password to use in the basic auth
public InfluxDBClient(string url, string username, string password) :
this(new InfluxDBClientOptions(url)
{
Username = username,
Password = password
})
{
}
///
/// Create a instance of the InfluxDB 2.x client.
///
/// the url to connect to the InfluxDB 2.x
/// the token to use for the authorization
public InfluxDBClient(string url, string token) :
this(new InfluxDBClientOptions(url)
{
Token = token
})
{
}
///
/// Create a instance of the InfluxDB 2.x client to connect into InfluxDB 1.8.
///
/// the url to connect to the InfluxDB 1.8
/// authorization username
/// authorization password
/// database name
/// retention policy
public InfluxDBClient(string url, string username, string password, string database, string retentionPolicy) :
this(new InfluxDBClientOptions(url)
{
Org = "-",
Token = $"{username}:{password}",
Bucket = $"{database}/{retentionPolicy}"
})
{
}
///
/// Create a instance of the InfluxDB 2.x client.
///
/// the connection configuration
public InfluxDBClient(InfluxDBClientOptions options)
{
Arguments.CheckNotNull(options, nameof(options));
_options = options;
_loggingHandler = new LoggingHandler(options.LogLevel);
_gzipHandler = new GzipHandler();
_apiClient = new ApiClient(options, _loggingHandler, _gzipHandler);
_exceptionFactory = (methodName, response) =>
!response.IsSuccessful ? HttpException.Create(response, response.Content) : null;
_setupService = new SetupService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
_healthService = new HealthService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
_readyService = new ReadyService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
_pingService = new PingService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
}
public void Dispose()
{
//
// Dispose child APIs
//
_disposeNotification.OnNext(Unit.Default);
//
// signout
//
try
{
_apiClient.Signout();
}
catch (Exception e)
{
Trace.WriteLine("The signout exception");
Trace.WriteLine(e);
}
//
// Dispose HttpClient
//
_apiClient.RestClient.Dispose();
}
///
/// Get the Query client.
///
/// the mapper used for mapping FluxResults to POCO
/// the new client instance for the Query API
IQueryApi IInfluxDBClient.GetQueryApi(IDomainObjectMapper mapper)
{
return GetQueryApi(mapper);
}
///
/// Get the Query client.
///
/// the mapper used for mapping FluxResults to POCO
/// the new client instance for the Query API
public QueryApi GetQueryApi(IDomainObjectMapper mapper = null)
{
var service = new QueryService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new QueryApi(_options, service, mapper ?? new DefaultDomainObjectMapper());
}
///
/// Get the synchronous version of Query client.
///
/// the mapper used for mapping FluxResults to POCO
/// the new synchronous client instance for the Query API
IQueryApiSync IInfluxDBClient.GetQueryApiSync(IDomainObjectMapper mapper)
{
return GetQueryApiSync(mapper);
}
///
/// Get the synchronous version of Query client.
///
/// the mapper used for mapping FluxResults to POCO
/// the new synchronous client instance for the Query API
public QueryApiSync GetQueryApiSync(IDomainObjectMapper mapper = null)
{
var service = new QueryService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new QueryApiSync(_options, service, mapper ?? new DefaultDomainObjectMapper());
}
///
/// Get the Write client.
///
/// the mapper used for mapping to PointData
/// the new client instance for the Write API
IWriteApi IInfluxDBClient.GetWriteApi(IDomainObjectMapper mapper)
{
return GetWriteApi(mapper);
}
///
/// Get the Write client.
///
/// the mapper used for mapping to PointData
/// the new client instance for the Write API
public WriteApi GetWriteApi(IDomainObjectMapper mapper = null)
{
return GetWriteApi(new WriteOptions(), mapper);
}
///
/// Get the Write async client.
///
/// the converter used for mapping to PointData
/// the new client instance for the Write API Async without batching
IWriteApiAsync IInfluxDBClient.GetWriteApiAsync(IDomainObjectMapper mapper)
{
return GetWriteApiAsync(mapper);
}
///
/// Get the Write async client.
///
/// the converter used for mapping to PointData
/// the new client instance for the Write API Async without batching
public WriteApiAsync GetWriteApiAsync(IDomainObjectMapper mapper = null)
{
var service = new WriteService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new WriteApiAsync(_options, service, mapper ?? new DefaultDomainObjectMapper(), this);
}
///
/// Get the Write client.
///
/// the configuration for a write client
/// the converter used for mapping to PointData
/// the new client instance for the Write API
IWriteApi IInfluxDBClient.GetWriteApi(WriteOptions writeOptions, IDomainObjectMapper mapper)
{
return GetWriteApi(writeOptions, mapper);
}
///
/// Get the Write client.
///
/// the configuration for a write client
/// the converter used for mapping to PointData
/// the new client instance for the Write API
public WriteApi GetWriteApi(WriteOptions writeOptions, IDomainObjectMapper mapper = null)
{
var service = new WriteService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
var writeApi = new WriteApi(_options, service, writeOptions, mapper ?? new DefaultDomainObjectMapper(),
this, _disposeNotification);
return writeApi;
}
///
/// Get the client.
///
/// the new client instance for Organization API
IOrganizationsApi IInfluxDBClient.GetOrganizationsApi()
{
return GetOrganizationsApi();
}
///
/// Get the client.
///
/// the new client instance for Organization API
public OrganizationsApi GetOrganizationsApi()
{
var service = new OrganizationsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
var secretService = new SecretsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new OrganizationsApi(service, secretService);
}
///
/// Get the client.
///
/// the new client instance for User API
IUsersApi IInfluxDBClient.GetUsersApi()
{
return GetUsersApi();
}
///
/// Get the client.
///
/// the new client instance for User API
public UsersApi GetUsersApi()
{
var service = new UsersService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new UsersApi(service);
}
///
/// Get the client.
///
/// the new client instance for Bucket API
IBucketsApi IInfluxDBClient.GetBucketsApi()
{
return GetBucketsApi();
}
///
/// Get the client.
///
/// the new client instance for Bucket API
public BucketsApi GetBucketsApi()
{
var service = new BucketsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new BucketsApi(service);
}
///
/// Get the client.
///
/// the new client instance for Source API
ISourcesApi IInfluxDBClient.GetSourcesApi()
{
return GetSourcesApi();
}
///
/// Get the client.
///
/// the new client instance for Source API
public SourcesApi GetSourcesApi()
{
var service = new SourcesService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new SourcesApi(service);
}
///
/// Get the client.
///
/// the new client instance for Authorization API
IAuthorizationsApi IInfluxDBClient.GetAuthorizationsApi()
{
return GetAuthorizationsApi();
}
///
/// Get the client.
///
/// the new client instance for Authorization API
public AuthorizationsApi GetAuthorizationsApi()
{
var service = new AuthorizationsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new AuthorizationsApi(service);
}
///
/// Get the client.
///
/// the new client instance for Task API
ITasksApi IInfluxDBClient.GetTasksApi()
{
return GetTasksApi();
}
///
/// Get the client.
///
/// the new client instance for Task API
public TasksApi GetTasksApi()
{
var service = new TasksService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new TasksApi(service);
}
///
/// Get the client.
///
/// the new client instance for Scraper API
IScraperTargetsApi IInfluxDBClient.GetScraperTargetsApi()
{
return GetScraperTargetsApi();
}
///
/// Get the client.
///
/// the new client instance for Scraper API
public ScraperTargetsApi GetScraperTargetsApi()
{
var service = new ScraperTargetsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new ScraperTargetsApi(service);
}
///
/// Get the client.
///
/// the new client instance for Telegrafs API
ITelegrafsApi IInfluxDBClient.GetTelegrafsApi()
{
return GetTelegrafsApi();
}
///
/// Get the client.
///
/// the new client instance for Telegrafs API
public TelegrafsApi GetTelegrafsApi()
{
var service = new TelegrafsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new TelegrafsApi(service);
}
///
/// Get the client.
///
/// the new client instance for Label API
ILabelsApi IInfluxDBClient.GetLabelsApi()
{
return GetLabelsApi();
}
///
/// Get the client.
///
/// the new client instance for Label API
public LabelsApi GetLabelsApi()
{
var service = new LabelsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new LabelsApi(service);
}
///
/// Get the client.
///
/// the new client instance for NotificationEndpoint API
INotificationEndpointsApi IInfluxDBClient.GetNotificationEndpointsApi()
{
return GetNotificationEndpointsApi();
}
///
/// Get the client.
///
/// the new client instance for NotificationEndpoint API
public NotificationEndpointsApi GetNotificationEndpointsApi()
{
var service = new NotificationEndpointsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new NotificationEndpointsApi(service);
}
///
/// Get the client.
///
/// the new client instance for NotificationRules API
INotificationRulesApi IInfluxDBClient.GetNotificationRulesApi()
{
return GetNotificationRulesApi();
}
///
/// Get the client.
///
/// the new client instance for NotificationRules API
public NotificationRulesApi GetNotificationRulesApi()
{
var service = new NotificationRulesService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new NotificationRulesApi(service);
}
///
/// Get the client.
///
/// the new client instance for Checks API
IChecksApi IInfluxDBClient.GetChecksApi()
{
return GetChecksApi();
}
///
/// Get the client.
///
/// the new client instance for Checks API
public ChecksApi GetChecksApi()
{
var service = new ChecksService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new ChecksApi(service);
}
///
/// Get the Delete client.
///
/// the new client instance for Delete API
IDeleteApi IInfluxDBClient.GetDeleteApi()
{
return GetDeleteApi();
}
///
/// Get the Delete client.
///
/// the new client instance for Delete API
public DeleteApi GetDeleteApi()
{
var service = new DeleteService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new DeleteApi(service);
}
///
/// Create an InvokableScripts API instance.
///
/// the mapper used for mapping invocation results to POCO
/// New instance of InvokableScriptsApi.
IInvokableScriptsApi IInfluxDBClient.GetInvokableScriptsApi(IDomainObjectMapper mapper)
{
return GetInvokableScriptsApi(mapper);
}
///
/// Create an InvokableScripts API instance.
///
/// the mapper used for mapping invocation results to POCO
/// New instance of InvokableScriptsApi.
public InvokableScriptsApi GetInvokableScriptsApi(IDomainObjectMapper mapper = null)
{
var service = new InvokableScriptsService((Configuration)_apiClient.Configuration)
{
ExceptionFactory = _exceptionFactory
};
return new InvokableScriptsApi(service, mapper ?? new DefaultDomainObjectMapper());
}
///
/// Create a service for specified type.
///
/// type of service
/// type of service
/// new instance of service
public TS CreateService(Type serviceType) where TS : IApiAccessor
{
var instance = (TS)Activator.CreateInstance(serviceType, (Configuration)_apiClient.Configuration);
instance.ExceptionFactory = _exceptionFactory;
return instance;
}
///
/// Set the log level for the request and response information.
///
/// the log level to set
public void SetLogLevel(LogLevel logLevel)
{
Arguments.CheckNotNull(logLevel, nameof(logLevel));
_loggingHandler.Level = logLevel;
}
///
/// Set the that is used for logging requests and responses.
///
/// Log Level
public LogLevel GetLogLevel()
{
return _loggingHandler.Level;
}
///
/// Enable Gzip compress for http requests.
///
/// Currently only the "Write" and "Query" endpoints supports the Gzip compression.
///
///
IInfluxDBClient IInfluxDBClient.EnableGzip()
{
return EnableGzip();
}
///
/// Enable Gzip compress for http requests.
///
/// Currently only the "Write" and "Query" endpoints supports the Gzip compression.
///
///
public InfluxDBClient EnableGzip()
{
_gzipHandler.EnableGzip();
return this;
}
///
/// Disable Gzip compress for http request body.
///
/// this
IInfluxDBClient IInfluxDBClient.DisableGzip()
{
return DisableGzip();
}
///
/// Disable Gzip compress for http request body.
///
/// this
public InfluxDBClient DisableGzip()
{
_gzipHandler.DisableGzip();
return this;
}
///
/// Returns whether Gzip compress for http request body is enabled.
///
/// true if gzip is enabled.
public bool IsGzipEnabled()
{
return _gzipHandler.IsEnabledGzip();
}
///
/// Get the health of an instance.
///
/// health of an instance
[Obsolete("This method is obsolete. Call 'PingAsync()' instead.", false)]
public Task HealthAsync()
{
return GetHealthAsync(_healthService.GetHealthAsync());
}
///
/// Check the status of InfluxDB Server.
///
/// true if server is healthy otherwise return false
public async Task PingAsync()
{
return await PingAsync(_pingService.GetPingAsyncWithIRestResponse());
}
///
/// Return the version of the connected InfluxDB Server.
///
/// the version String, otherwise unknown
/// throws when request did not succesfully ends
public async Task VersionAsync()
{
return await VersionAsync(_pingService.GetPingAsyncWithIRestResponse());
}
///
/// Check the readiness of InfluxDB Server at startup. It is not supported by InfluxDB Cloud.
///
/// return null if the InfluxDB is not ready
public async Task ReadyAsync()
{
try
{
return await _readyService.GetReadyAsync().ConfigureAwait(false);
}
catch (Exception e)
{
Trace.TraceError($"The exception: '{e.Message}' occurs during check instance readiness.");
return null;
}
}
///
/// Post onboarding request, to setup initial user, org and bucket.
///
/// to setup defaults
/// With status code 422 when an onboarding has already been completed
/// defaults for first run
public Task OnboardingAsync(OnboardingRequest onboarding)
{
Arguments.CheckNotNull(onboarding, nameof(onboarding));
return _setupService.PostSetupAsync(onboarding);
}
///
/// Check if database has default user, org, bucket created, returns true if not.
///
/// True if onboarding has already been completed otherwise false
public async Task IsOnboardingAllowedAsync()
{
var isOnboarding = await _setupService.GetSetupAsync().ConfigureAwait(false);
return isOnboarding.Allowed == true;
}
internal static string AuthorizationHeader(string username, string password)
{
return "Basic " + Convert.ToBase64String(Encoding.Default.GetBytes(username + ":" + password));
}
internal static async Task GetHealthAsync(Task task)
{
Arguments.CheckNotNull(task, nameof(task));
try
{
return await task.ConfigureAwait(false);
}
catch (Exception e)
{
return new HealthCheck("influxdb", e.Message, default, HealthCheck.StatusEnum.Fail);
}
}
}
}