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); } } } }