From 82225075ecd4d16ec4165d10765056911ef768fb Mon Sep 17 00:00:00 2001 From: AnastasiiaKomar Date: Thu, 21 Jul 2022 17:41:27 +0300 Subject: [PATCH] http sending + checksum improved --- .../sql/schema-entities-idx-psql-addon.sql | 38 + .../src/main/data/sql/schema-entities-idx.sql | 53 ++ .../src/main/data/sql/schema-entities.sql | 695 ++++++++++++++++++ .../src/main/data/sql/schema-timescale.sql | 157 ++++ .../src/main/data/sql/schema-ts-psql.sql | 339 +++++++++ .../controller/OtaPackageController.java | 9 +- .../src/main/resources/thingsboard.yml | 10 +- .../cache/ota/CaffeineOtaPackageCache.java | 8 +- .../cache/ota/service/FileCacheService.java | 148 ++-- .../server/dao/ota/OtaPackageService.java | 11 +- .../common/data/ota/ChecksumAlgorithm.java | 31 +- .../server/dao/ota/BaseOtaPackageService.java | 73 +- .../server/dao/ota/util/ChecksumUtil.java | 79 ++ .../validator/OtaPackageDataValidator.java | 7 +- .../server/dao/ota/util/ChecksumUtilTest.java | 58 ++ .../dao/service/AbstractServiceTest.java | 3 +- .../service/BaseDeviceProfileServiceTest.java | 3 +- .../dao/service/BaseDeviceServiceTest.java | 5 +- .../service/BaseOtaPackageServiceTest.java | 4 +- 19 files changed, 1571 insertions(+), 160 deletions(-) create mode 100644 application/src/main/data/sql/schema-entities-idx-psql-addon.sql create mode 100644 application/src/main/data/sql/schema-entities-idx.sql create mode 100644 application/src/main/data/sql/schema-entities.sql create mode 100644 application/src/main/data/sql/schema-timescale.sql create mode 100644 application/src/main/data/sql/schema-ts-psql.sql create mode 100644 dao/src/main/java/org/thingsboard/server/dao/ota/util/ChecksumUtil.java create mode 100644 dao/src/test/java/org/thingsboard/server/dao/ota/util/ChecksumUtilTest.java diff --git a/application/src/main/data/sql/schema-entities-idx-psql-addon.sql b/application/src/main/data/sql/schema-entities-idx-psql-addon.sql new file mode 100644 index 0000000000..7e15b419a8 --- /dev/null +++ b/application/src/main/data/sql/schema-entities-idx-psql-addon.sql @@ -0,0 +1,38 @@ +-- +-- Copyright © 2016-2022 The Thingsboard Authors +-- +-- Licensed 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. +-- + +-- This file describes PostgreSQL-specific indexes that not supported by hsql +-- It is not a stand-alone file! Run schema-entities-idx.sql before! +-- Note: Hibernate DESC order translates to native SQL "ORDER BY .. DESC NULLS LAST" +-- While creating index PostgreSQL transforms short notation (ts DESC) to the full (DESC NULLS FIRST) +-- That difference between NULLS LAST and NULLS FIRST prevents to hit index while querying latest by ts +-- That why we need to define DESC index explicitly as (ts DESC NULLS LAST) + +CREATE INDEX IF NOT EXISTS idx_event_ts + ON public.event + (ts DESC NULLS LAST) + WITH (FILLFACTOR=95); + +COMMENT ON INDEX public.idx_event_ts + IS 'This index helps to delete events by TTL using timestamp'; + +CREATE INDEX IF NOT EXISTS idx_event_tenant_entity_type_entity_event_type_created_time_des + ON public.event + (tenant_id ASC, entity_type ASC, entity_id ASC, event_type ASC, created_time DESC NULLS LAST) + WITH (FILLFACTOR=95); + +COMMENT ON INDEX public.idx_event_tenant_entity_type_entity_event_type_created_time_des + IS 'This index helps to open latest events on UI fast'; diff --git a/application/src/main/data/sql/schema-entities-idx.sql b/application/src/main/data/sql/schema-entities-idx.sql new file mode 100644 index 0000000000..b74d935c12 --- /dev/null +++ b/application/src/main/data/sql/schema-entities-idx.sql @@ -0,0 +1,53 @@ +-- +-- Copyright © 2016-2022 The Thingsboard Authors +-- +-- Licensed 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. +-- + +CREATE INDEX IF NOT EXISTS idx_alarm_originator_alarm_type ON alarm(originator_id, type, start_ts DESC); + +CREATE INDEX IF NOT EXISTS idx_alarm_originator_created_time ON alarm(originator_id, created_time DESC); + +CREATE INDEX IF NOT EXISTS idx_alarm_tenant_created_time ON alarm(tenant_id, created_time DESC); + +CREATE INDEX IF NOT EXISTS idx_alarm_tenant_status_created_time ON alarm(tenant_id, status, created_time DESC); + +CREATE INDEX IF NOT EXISTS idx_alarm_tenant_alarm_type_created_time ON alarm(tenant_id, type, created_time DESC); + +CREATE INDEX IF NOT EXISTS idx_entity_alarm_created_time ON entity_alarm(tenant_id, entity_id, created_time DESC); + +CREATE INDEX IF NOT EXISTS idx_entity_alarm_alarm_id ON entity_alarm(alarm_id); + +CREATE INDEX IF NOT EXISTS idx_relation_to_id ON relation(relation_type_group, to_type, to_id); + +CREATE INDEX IF NOT EXISTS idx_relation_from_id ON relation(relation_type_group, from_type, from_id); + +CREATE INDEX IF NOT EXISTS idx_device_customer_id ON device(tenant_id, customer_id); + +CREATE INDEX IF NOT EXISTS idx_device_customer_id_and_type ON device(tenant_id, customer_id, type); + +CREATE INDEX IF NOT EXISTS idx_device_type ON device(tenant_id, type); + +CREATE INDEX IF NOT EXISTS idx_device_device_profile_id ON device(tenant_id, device_profile_id); + +CREATE INDEX IF NOT EXISTS idx_asset_customer_id ON asset(tenant_id, customer_id); + +CREATE INDEX IF NOT EXISTS idx_asset_customer_id_and_type ON asset(tenant_id, customer_id, type); + +CREATE INDEX IF NOT EXISTS idx_asset_type ON asset(tenant_id, type); + +CREATE INDEX IF NOT EXISTS idx_attribute_kv_by_key_and_last_update_ts ON attribute_kv(entity_id, attribute_key, last_update_ts desc); + +CREATE INDEX IF NOT EXISTS idx_audit_log_tenant_id_and_created_time ON audit_log(tenant_id, created_time); + +CREATE INDEX IF NOT EXISTS idx_rpc_tenant_id_device_id ON rpc(tenant_id, device_id); diff --git a/application/src/main/data/sql/schema-entities.sql b/application/src/main/data/sql/schema-entities.sql new file mode 100644 index 0000000000..4e2596485b --- /dev/null +++ b/application/src/main/data/sql/schema-entities.sql @@ -0,0 +1,695 @@ +-- +-- Copyright © 2016-2022 The Thingsboard Authors +-- +-- Licensed 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. +-- + +CREATE TABLE IF NOT EXISTS tb_schema_settings +( + schema_version bigint NOT NULL, + CONSTRAINT tb_schema_settings_pkey PRIMARY KEY (schema_version) +); + +CREATE OR REPLACE PROCEDURE insert_tb_schema_settings() + LANGUAGE plpgsql AS +$$ +BEGIN + IF (SELECT COUNT(*) FROM tb_schema_settings) = 0 THEN + INSERT INTO tb_schema_settings (schema_version) VALUES (3003000); + END IF; +END; +$$; + +call insert_tb_schema_settings(); + +CREATE TABLE IF NOT EXISTS admin_settings ( + id uuid NOT NULL CONSTRAINT admin_settings_pkey PRIMARY KEY, + created_time bigint NOT NULL, + json_value varchar, + key varchar(255) +); + +CREATE TABLE IF NOT EXISTS alarm ( + id uuid NOT NULL CONSTRAINT alarm_pkey PRIMARY KEY, + created_time bigint NOT NULL, + ack_ts bigint, + clear_ts bigint, + additional_info varchar, + end_ts bigint, + originator_id uuid, + originator_type integer, + propagate boolean, + severity varchar(255), + start_ts bigint, + status varchar(255), + tenant_id uuid, + customer_id uuid, + propagate_relation_types varchar, + type varchar(255), + propagate_to_owner boolean, + propagate_to_tenant boolean +); + +CREATE TABLE IF NOT EXISTS entity_alarm ( + tenant_id uuid NOT NULL, + entity_type varchar(32), + entity_id uuid NOT NULL, + created_time bigint NOT NULL, + alarm_type varchar(255) NOT NULL, + customer_id uuid, + alarm_id uuid, + CONSTRAINT entity_alarm_pkey PRIMARY KEY (entity_id, alarm_id), + CONSTRAINT fk_entity_alarm_id FOREIGN KEY (alarm_id) REFERENCES alarm(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS asset ( + id uuid NOT NULL CONSTRAINT asset_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + customer_id uuid, + name varchar(255), + label varchar(255), + search_text varchar(255), + tenant_id uuid, + type varchar(255), + CONSTRAINT asset_name_unq_key UNIQUE (tenant_id, name) +); + +CREATE TABLE IF NOT EXISTS audit_log ( + id uuid NOT NULL CONSTRAINT audit_log_pkey PRIMARY KEY, + created_time bigint NOT NULL, + tenant_id uuid, + customer_id uuid, + entity_id uuid, + entity_type varchar(255), + entity_name varchar(255), + user_id uuid, + user_name varchar(255), + action_type varchar(255), + action_data varchar(1000000), + action_status varchar(255), + action_failure_details varchar(1000000) +); + +CREATE TABLE IF NOT EXISTS attribute_kv ( + entity_type varchar(255), + entity_id uuid, + attribute_type varchar(255), + attribute_key varchar(255), + bool_v boolean, + str_v varchar(10000000), + long_v bigint, + dbl_v double precision, + json_v json, + last_update_ts bigint, + CONSTRAINT attribute_kv_pkey PRIMARY KEY (entity_type, entity_id, attribute_type, attribute_key) +); + +CREATE TABLE IF NOT EXISTS component_descriptor ( + id uuid NOT NULL CONSTRAINT component_descriptor_pkey PRIMARY KEY, + created_time bigint NOT NULL, + actions varchar(255), + clazz varchar UNIQUE, + configuration_descriptor varchar, + name varchar(255), + scope varchar(255), + search_text varchar(255), + type varchar(255) +); + +CREATE TABLE IF NOT EXISTS customer ( + id uuid NOT NULL CONSTRAINT customer_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + address varchar, + address2 varchar, + city varchar(255), + country varchar(255), + email varchar(255), + phone varchar(255), + search_text varchar(255), + state varchar(255), + tenant_id uuid, + title varchar(255), + zip varchar(255) +); + +CREATE TABLE IF NOT EXISTS dashboard ( + id uuid NOT NULL CONSTRAINT dashboard_pkey PRIMARY KEY, + created_time bigint NOT NULL, + configuration varchar, + assigned_customers varchar(1000000), + search_text varchar(255), + tenant_id uuid, + title varchar(255), + mobile_hide boolean DEFAULT false, + mobile_order int, + image varchar(1000000) +); + +CREATE TABLE IF NOT EXISTS rule_chain ( + id uuid NOT NULL CONSTRAINT rule_chain_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + configuration varchar(10000000), + name varchar(255), + type varchar(255), + first_rule_node_id uuid, + root boolean, + debug_mode boolean, + search_text varchar(255), + tenant_id uuid +); + +CREATE TABLE IF NOT EXISTS rule_node ( + id uuid NOT NULL CONSTRAINT rule_node_pkey PRIMARY KEY, + created_time bigint NOT NULL, + rule_chain_id uuid, + additional_info varchar, + configuration varchar(10000000), + type varchar(255), + name varchar(255), + debug_mode boolean, + search_text varchar(255) +); + +CREATE TABLE IF NOT EXISTS rule_node_state ( + id uuid NOT NULL CONSTRAINT rule_node_state_pkey PRIMARY KEY, + created_time bigint NOT NULL, + rule_node_id uuid NOT NULL, + entity_type varchar(32) NOT NULL, + entity_id uuid NOT NULL, + state_data varchar(16384) NOT NULL, + CONSTRAINT rule_node_state_unq_key UNIQUE (rule_node_id, entity_id), + CONSTRAINT fk_rule_node_state_node_id FOREIGN KEY (rule_node_id) REFERENCES rule_node(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS ota_package ( + id uuid NOT NULL CONSTRAINT ota_package_pkey PRIMARY KEY, + created_time bigint NOT NULL, + tenant_id uuid NOT NULL, + device_profile_id uuid , + type varchar(32) NOT NULL, + title varchar(255) NOT NULL, + version varchar(255) NOT NULL, + tag varchar(255), + url varchar(255), + file_name varchar(255), + content_type varchar(255), + checksum_algorithm varchar(32), + checksum varchar(1020), + data oid, + data_size bigint, + additional_info varchar, + search_text varchar(255), + CONSTRAINT ota_package_tenant_title_version_unq_key UNIQUE (tenant_id, title, version) +); + +CREATE TABLE IF NOT EXISTS device_profile ( + id uuid NOT NULL CONSTRAINT device_profile_pkey PRIMARY KEY, + created_time bigint NOT NULL, + name varchar(255), + type varchar(255), + image varchar(1000000), + transport_type varchar(255), + provision_type varchar(255), + profile_data jsonb, + description varchar, + search_text varchar(255), + is_default boolean, + tenant_id uuid, + firmware_id uuid, + software_id uuid, + default_rule_chain_id uuid, + default_dashboard_id uuid, + default_queue_name varchar(255), + provision_device_key varchar, + CONSTRAINT device_profile_name_unq_key UNIQUE (tenant_id, name), + CONSTRAINT device_provision_key_unq_key UNIQUE (provision_device_key), + CONSTRAINT fk_default_rule_chain_device_profile FOREIGN KEY (default_rule_chain_id) REFERENCES rule_chain(id), + CONSTRAINT fk_default_dashboard_device_profile FOREIGN KEY (default_dashboard_id) REFERENCES dashboard(id), + CONSTRAINT fk_firmware_device_profile FOREIGN KEY (firmware_id) REFERENCES ota_package(id), + CONSTRAINT fk_software_device_profile FOREIGN KEY (software_id) REFERENCES ota_package(id) +); + +ALTER TABLE ota_package + ADD CONSTRAINT fk_device_profile_ota_package + FOREIGN KEY (device_profile_id) REFERENCES device_profile (id) + ON DELETE CASCADE; + +-- We will use one-to-many relation in the first release and extend this feature in case of user requests +-- CREATE TABLE IF NOT EXISTS device_profile_firmware ( +-- device_profile_id uuid NOT NULL, +-- firmware_id uuid NOT NULL, +-- CONSTRAINT device_profile_firmware_unq_key UNIQUE (device_profile_id, firmware_id), +-- CONSTRAINT fk_device_profile_firmware_device_profile FOREIGN KEY (device_profile_id) REFERENCES device_profile(id) ON DELETE CASCADE, +-- CONSTRAINT fk_device_profile_firmware_firmware FOREIGN KEY (firmware_id) REFERENCES firmware(id) ON DELETE CASCADE, +-- ); + +CREATE TABLE IF NOT EXISTS device ( + id uuid NOT NULL CONSTRAINT device_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + customer_id uuid, + device_profile_id uuid NOT NULL, + device_data jsonb, + type varchar(255), + name varchar(255), + label varchar(255), + search_text varchar(255), + tenant_id uuid, + firmware_id uuid, + software_id uuid, + CONSTRAINT device_name_unq_key UNIQUE (tenant_id, name), + CONSTRAINT fk_device_profile FOREIGN KEY (device_profile_id) REFERENCES device_profile(id), + CONSTRAINT fk_firmware_device FOREIGN KEY (firmware_id) REFERENCES ota_package(id), + CONSTRAINT fk_software_device FOREIGN KEY (software_id) REFERENCES ota_package(id) +); + +CREATE TABLE IF NOT EXISTS device_credentials ( + id uuid NOT NULL CONSTRAINT device_credentials_pkey PRIMARY KEY, + created_time bigint NOT NULL, + credentials_id varchar, + credentials_type varchar(255), + credentials_value varchar, + device_id uuid, + CONSTRAINT device_credentials_id_unq_key UNIQUE (credentials_id), + CONSTRAINT device_credentials_device_id_unq_key UNIQUE (device_id) +); + +CREATE TABLE IF NOT EXISTS event ( + id uuid NOT NULL CONSTRAINT event_pkey PRIMARY KEY, + created_time bigint NOT NULL, + body varchar(10000000), + entity_id uuid, + entity_type varchar(255), + event_type varchar(255), + event_uid varchar(255), + tenant_id uuid, + ts bigint NOT NULL, + CONSTRAINT event_unq_key UNIQUE (tenant_id, entity_type, entity_id, event_type, event_uid) +); + +CREATE TABLE IF NOT EXISTS relation ( + from_id uuid, + from_type varchar(255), + to_id uuid, + to_type varchar(255), + relation_type_group varchar(255), + relation_type varchar(255), + additional_info varchar, + CONSTRAINT relation_pkey PRIMARY KEY (from_id, from_type, relation_type_group, relation_type, to_id, to_type) +); +-- ) PARTITION BY LIST (relation_type_group); +-- +-- CREATE TABLE other_relations PARTITION OF relation DEFAULT; +-- CREATE TABLE common_relations PARTITION OF relation FOR VALUES IN ('COMMON'); +-- CREATE TABLE alarm_relations PARTITION OF relation FOR VALUES IN ('ALARM'); +-- CREATE TABLE dashboard_relations PARTITION OF relation FOR VALUES IN ('DASHBOARD'); +-- CREATE TABLE rule_relations PARTITION OF relation FOR VALUES IN ('RULE_CHAIN', 'RULE_NODE'); + +CREATE TABLE IF NOT EXISTS tb_user ( + id uuid NOT NULL CONSTRAINT tb_user_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + authority varchar(255), + customer_id uuid, + email varchar(255) UNIQUE, + first_name varchar(255), + last_name varchar(255), + search_text varchar(255), + tenant_id uuid +); + +CREATE TABLE IF NOT EXISTS tenant_profile ( + id uuid NOT NULL CONSTRAINT tenant_profile_pkey PRIMARY KEY, + created_time bigint NOT NULL, + name varchar(255), + profile_data jsonb, + description varchar, + search_text varchar(255), + is_default boolean, + isolated_tb_core boolean, + isolated_tb_rule_engine boolean, + CONSTRAINT tenant_profile_name_unq_key UNIQUE (name) +); + +CREATE TABLE IF NOT EXISTS tenant ( + id uuid NOT NULL CONSTRAINT tenant_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + tenant_profile_id uuid NOT NULL, + address varchar, + address2 varchar, + city varchar(255), + country varchar(255), + email varchar(255), + phone varchar(255), + region varchar(255), + search_text varchar(255), + state varchar(255), + title varchar(255), + zip varchar(255), + CONSTRAINT fk_tenant_profile FOREIGN KEY (tenant_profile_id) REFERENCES tenant_profile(id) +); + +CREATE TABLE IF NOT EXISTS user_credentials ( + id uuid NOT NULL CONSTRAINT user_credentials_pkey PRIMARY KEY, + created_time bigint NOT NULL, + activate_token varchar(255) UNIQUE, + enabled boolean, + password varchar(255), + reset_token varchar(255) UNIQUE, + user_id uuid UNIQUE +); + +CREATE TABLE IF NOT EXISTS widget_type ( + id uuid NOT NULL CONSTRAINT widget_type_pkey PRIMARY KEY, + created_time bigint NOT NULL, + alias varchar(255), + bundle_alias varchar(255), + descriptor varchar(1000000), + name varchar(255), + tenant_id uuid, + image varchar(1000000), + description varchar(255) +); + +CREATE TABLE IF NOT EXISTS widgets_bundle ( + id uuid NOT NULL CONSTRAINT widgets_bundle_pkey PRIMARY KEY, + created_time bigint NOT NULL, + alias varchar(255), + search_text varchar(255), + tenant_id uuid, + title varchar(255), + image varchar(1000000), + description varchar(255) +); + +CREATE TABLE IF NOT EXISTS entity_view ( + id uuid NOT NULL CONSTRAINT entity_view_pkey PRIMARY KEY, + created_time bigint NOT NULL, + entity_id uuid, + entity_type varchar(255), + tenant_id uuid, + customer_id uuid, + type varchar(255), + name varchar(255), + keys varchar(10000000), + start_ts bigint, + end_ts bigint, + search_text varchar(255), + additional_info varchar +); + +CREATE TABLE IF NOT EXISTS ts_kv_latest +( + entity_id uuid NOT NULL, + key int NOT NULL, + ts bigint NOT NULL, + bool_v boolean, + str_v varchar(10000000), + long_v bigint, + dbl_v double precision, + json_v json, + CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_id, key) +); + +CREATE TABLE IF NOT EXISTS ts_kv_dictionary +( + key varchar(255) NOT NULL, + key_id serial UNIQUE, + CONSTRAINT ts_key_id_pkey PRIMARY KEY (key) +); + +CREATE TABLE IF NOT EXISTS oauth2_params ( + id uuid NOT NULL CONSTRAINT oauth2_params_pkey PRIMARY KEY, + enabled boolean, + tenant_id uuid, + created_time bigint NOT NULL +); + +CREATE TABLE IF NOT EXISTS oauth2_registration ( + id uuid NOT NULL CONSTRAINT oauth2_registration_pkey PRIMARY KEY, + oauth2_params_id uuid NOT NULL, + created_time bigint NOT NULL, + additional_info varchar, + client_id varchar(255), + client_secret varchar(2048), + authorization_uri varchar(255), + token_uri varchar(255), + scope varchar(255), + platforms varchar(255), + user_info_uri varchar(255), + user_name_attribute_name varchar(255), + jwk_set_uri varchar(255), + client_authentication_method varchar(255), + login_button_label varchar(255), + login_button_icon varchar(255), + allow_user_creation boolean, + activate_user boolean, + type varchar(31), + basic_email_attribute_key varchar(31), + basic_first_name_attribute_key varchar(31), + basic_last_name_attribute_key varchar(31), + basic_tenant_name_strategy varchar(31), + basic_tenant_name_pattern varchar(255), + basic_customer_name_pattern varchar(255), + basic_default_dashboard_name varchar(255), + basic_always_full_screen boolean, + custom_url varchar(255), + custom_username varchar(255), + custom_password varchar(255), + custom_send_token boolean, + CONSTRAINT fk_registration_oauth2_params FOREIGN KEY (oauth2_params_id) REFERENCES oauth2_params(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS oauth2_domain ( + id uuid NOT NULL CONSTRAINT oauth2_domain_pkey PRIMARY KEY, + oauth2_params_id uuid NOT NULL, + created_time bigint NOT NULL, + domain_name varchar(255), + domain_scheme varchar(31), + CONSTRAINT fk_domain_oauth2_params FOREIGN KEY (oauth2_params_id) REFERENCES oauth2_params(id) ON DELETE CASCADE, + CONSTRAINT oauth2_domain_unq_key UNIQUE (oauth2_params_id, domain_name, domain_scheme) +); + +CREATE TABLE IF NOT EXISTS oauth2_mobile ( + id uuid NOT NULL CONSTRAINT oauth2_mobile_pkey PRIMARY KEY, + oauth2_params_id uuid NOT NULL, + created_time bigint NOT NULL, + pkg_name varchar(255), + app_secret varchar(2048), + CONSTRAINT fk_mobile_oauth2_params FOREIGN KEY (oauth2_params_id) REFERENCES oauth2_params(id) ON DELETE CASCADE, + CONSTRAINT oauth2_mobile_unq_key UNIQUE (oauth2_params_id, pkg_name) +); + +CREATE TABLE IF NOT EXISTS oauth2_client_registration_template ( + id uuid NOT NULL CONSTRAINT oauth2_client_registration_template_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + provider_id varchar(255), + authorization_uri varchar(255), + token_uri varchar(255), + scope varchar(255), + user_info_uri varchar(255), + user_name_attribute_name varchar(255), + jwk_set_uri varchar(255), + client_authentication_method varchar(255), + type varchar(31), + basic_email_attribute_key varchar(31), + basic_first_name_attribute_key varchar(31), + basic_last_name_attribute_key varchar(31), + basic_tenant_name_strategy varchar(31), + basic_tenant_name_pattern varchar(255), + basic_customer_name_pattern varchar(255), + basic_default_dashboard_name varchar(255), + basic_always_full_screen boolean, + comment varchar, + login_button_icon varchar(255), + login_button_label varchar(255), + help_link varchar(255), + CONSTRAINT oauth2_template_provider_id_unq_key UNIQUE (provider_id) +); + +-- Deprecated +CREATE TABLE IF NOT EXISTS oauth2_client_registration_info ( + id uuid NOT NULL CONSTRAINT oauth2_client_registration_info_pkey PRIMARY KEY, + enabled boolean, + created_time bigint NOT NULL, + additional_info varchar, + client_id varchar(255), + client_secret varchar(255), + authorization_uri varchar(255), + token_uri varchar(255), + scope varchar(255), + user_info_uri varchar(255), + user_name_attribute_name varchar(255), + jwk_set_uri varchar(255), + client_authentication_method varchar(255), + login_button_label varchar(255), + login_button_icon varchar(255), + allow_user_creation boolean, + activate_user boolean, + type varchar(31), + basic_email_attribute_key varchar(31), + basic_first_name_attribute_key varchar(31), + basic_last_name_attribute_key varchar(31), + basic_tenant_name_strategy varchar(31), + basic_tenant_name_pattern varchar(255), + basic_customer_name_pattern varchar(255), + basic_default_dashboard_name varchar(255), + basic_always_full_screen boolean, + custom_url varchar(255), + custom_username varchar(255), + custom_password varchar(255), + custom_send_token boolean +); + +-- Deprecated +CREATE TABLE IF NOT EXISTS oauth2_client_registration ( + id uuid NOT NULL CONSTRAINT oauth2_client_registration_pkey PRIMARY KEY, + created_time bigint NOT NULL, + domain_name varchar(255), + domain_scheme varchar(31), + client_registration_info_id uuid +); + +CREATE TABLE IF NOT EXISTS api_usage_state ( + id uuid NOT NULL CONSTRAINT usage_record_pkey PRIMARY KEY, + created_time bigint NOT NULL, + tenant_id uuid, + entity_type varchar(32), + entity_id uuid, + transport varchar(32), + db_storage varchar(32), + re_exec varchar(32), + js_exec varchar(32), + email_exec varchar(32), + sms_exec varchar(32), + alarm_exec varchar(32), + CONSTRAINT api_usage_state_unq_key UNIQUE (tenant_id, entity_id) +); + +CREATE TABLE IF NOT EXISTS resource ( + id uuid NOT NULL CONSTRAINT resource_pkey PRIMARY KEY, + created_time bigint NOT NULL, + tenant_id uuid NOT NULL, + title varchar(255) NOT NULL, + resource_type varchar(32) NOT NULL, + resource_key varchar(255) NOT NULL, + search_text varchar(255), + file_name varchar(255) NOT NULL, + data varchar, + CONSTRAINT resource_unq_key UNIQUE (tenant_id, resource_type, resource_key) +); + +CREATE TABLE IF NOT EXISTS edge ( + id uuid NOT NULL CONSTRAINT edge_pkey PRIMARY KEY, + created_time bigint NOT NULL, + additional_info varchar, + customer_id uuid, + root_rule_chain_id uuid, + type varchar(255), + name varchar(255), + label varchar(255), + routing_key varchar(255), + secret varchar(255), + search_text varchar(255), + tenant_id uuid, + CONSTRAINT edge_name_unq_key UNIQUE (tenant_id, name), + CONSTRAINT edge_routing_key_unq_key UNIQUE (routing_key) +); + +CREATE TABLE IF NOT EXISTS edge_event ( + id uuid NOT NULL CONSTRAINT edge_event_pkey PRIMARY KEY, + created_time bigint NOT NULL, + edge_id uuid, + edge_event_type varchar(255), + edge_event_uid varchar(255), + entity_id uuid, + edge_event_action varchar(255), + body varchar(10000000), + tenant_id uuid, + ts bigint NOT NULL +); + +CREATE TABLE IF NOT EXISTS rpc ( + id uuid NOT NULL CONSTRAINT rpc_pkey PRIMARY KEY, + created_time bigint NOT NULL, + tenant_id uuid NOT NULL, + device_id uuid NOT NULL, + expiration_time bigint NOT NULL, + request varchar(10000000) NOT NULL, + response varchar(10000000), + additional_info varchar(10000000), + status varchar(255) NOT NULL +); + +CREATE OR REPLACE PROCEDURE cleanup_events_by_ttl( + IN regular_events_start_ts bigint, + IN regular_events_end_ts bigint, + IN debug_events_start_ts bigint, + IN debug_events_end_ts bigint, + INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + ttl_deleted_count bigint DEFAULT 0; + debug_ttl_deleted_count bigint DEFAULT 0; +BEGIN + IF regular_events_start_ts > 0 AND regular_events_end_ts > 0 THEN + EXECUTE format( + 'WITH deleted AS (DELETE FROM event WHERE id in (SELECT id from event WHERE ts > %L::bigint AND ts < %L::bigint AND ' || + '(event_type != %L::varchar AND event_type != %L::varchar)) RETURNING *) ' || + 'SELECT count(*) FROM deleted', regular_events_start_ts, regular_events_end_ts, + 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into ttl_deleted_count; + END IF; + IF debug_events_start_ts > 0 AND debug_events_end_ts > 0 THEN + EXECUTE format( + 'WITH deleted AS (DELETE FROM event WHERE id in (SELECT id from event WHERE ts > %L::bigint AND ts < %L::bigint AND ' || + '(event_type = %L::varchar OR event_type = %L::varchar)) RETURNING *) ' || + 'SELECT count(*) FROM deleted', debug_events_start_ts, debug_events_end_ts, + 'DEBUG_RULE_NODE', 'DEBUG_RULE_CHAIN') into debug_ttl_deleted_count; + END IF; + RAISE NOTICE 'Events removed by ttl: %', ttl_deleted_count; + RAISE NOTICE 'Debug Events removed by ttl: %', debug_ttl_deleted_count; + deleted := ttl_deleted_count + debug_ttl_deleted_count; +END +$$; + +CREATE OR REPLACE FUNCTION to_uuid(IN entity_id varchar, OUT uuid_id uuid) AS +$$ +BEGIN + uuid_id := substring(entity_id, 8, 8) || '-' || substring(entity_id, 4, 4) || '-1' || substring(entity_id, 1, 3) || + '-' || substring(entity_id, 16, 4) || '-' || substring(entity_id, 20, 12); +END; +$$ LANGUAGE plpgsql; + + +CREATE OR REPLACE PROCEDURE cleanup_edge_events_by_ttl(IN ttl bigint, INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + ttl_ts bigint; + ttl_deleted_count bigint DEFAULT 0; +BEGIN + IF ttl > 0 THEN + ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - ttl::bigint * 1000)::bigint; + EXECUTE format( + 'WITH deleted AS (DELETE FROM edge_event WHERE ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', ttl_ts) into ttl_deleted_count; + END IF; + RAISE NOTICE 'Edge events removed by ttl: %', ttl_deleted_count; + deleted := ttl_deleted_count; +END +$$; diff --git a/application/src/main/data/sql/schema-timescale.sql b/application/src/main/data/sql/schema-timescale.sql new file mode 100644 index 0000000000..81b71f1a03 --- /dev/null +++ b/application/src/main/data/sql/schema-timescale.sql @@ -0,0 +1,157 @@ +-- +-- Copyright © 2016-2022 The Thingsboard Authors +-- +-- Licensed 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. +-- + +CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE; + +CREATE TABLE IF NOT EXISTS ts_kv ( + entity_id uuid NOT NULL, + key int NOT NULL, + ts bigint NOT NULL, + bool_v boolean, + str_v varchar(10000000), + long_v bigint, + dbl_v double precision, + json_v json, + CONSTRAINT ts_kv_pkey PRIMARY KEY (entity_id, key, ts) +); + +CREATE TABLE IF NOT EXISTS ts_kv_dictionary ( + key varchar(255) NOT NULL, + key_id serial UNIQUE, + CONSTRAINT ts_key_id_pkey PRIMARY KEY (key) +); + +CREATE TABLE IF NOT EXISTS ts_kv_latest ( + entity_id uuid NOT NULL, + key int NOT NULL, + ts bigint NOT NULL, + bool_v boolean, + str_v varchar(10000000), + long_v bigint, + dbl_v double precision, + json_v json, + CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_id, key) +); + +CREATE OR REPLACE FUNCTION to_uuid(IN entity_id varchar, OUT uuid_id uuid) AS +$$ +BEGIN + uuid_id := substring(entity_id, 8, 8) || '-' || substring(entity_id, 4, 4) || '-1' || substring(entity_id, 1, 3) || + '-' || substring(entity_id, 16, 4) || '-' || substring(entity_id, 20, 12); +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION delete_device_records_from_ts_kv(tenant_id uuid, customer_id uuid, ttl bigint, + OUT deleted bigint) AS +$$ +BEGIN + EXECUTE format( + 'WITH deleted AS (DELETE FROM ts_kv WHERE entity_id IN (SELECT device.id as entity_id FROM device WHERE tenant_id = %L and customer_id = %L) AND ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', + tenant_id, customer_id, ttl) into deleted; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION delete_asset_records_from_ts_kv(tenant_id uuid, customer_id uuid, ttl bigint, + OUT deleted bigint) AS +$$ +BEGIN + EXECUTE format( + 'WITH deleted AS (DELETE FROM ts_kv WHERE entity_id IN (SELECT asset.id as entity_id FROM asset WHERE tenant_id = %L and customer_id = %L) AND ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', + tenant_id, customer_id, ttl) into deleted; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION delete_customer_records_from_ts_kv(tenant_id uuid, customer_id uuid, ttl bigint, + OUT deleted bigint) AS +$$ +BEGIN + EXECUTE format( + 'WITH deleted AS (DELETE FROM ts_kv WHERE entity_id IN (SELECT customer.id as entity_id FROM customer WHERE tenant_id = %L and id = %L) AND ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', + tenant_id, customer_id, ttl) into deleted; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE PROCEDURE cleanup_timeseries_by_ttl(IN null_uuid uuid, + IN system_ttl bigint, INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + tenant_cursor CURSOR FOR select tenant.id as tenant_id + from tenant; + tenant_id_record uuid; + customer_id_record uuid; + tenant_ttl bigint; + customer_ttl bigint; + deleted_for_entities bigint; + tenant_ttl_ts bigint; + customer_ttl_ts bigint; +BEGIN + OPEN tenant_cursor; + FETCH tenant_cursor INTO tenant_id_record; + WHILE FOUND + LOOP + EXECUTE format( + 'select attribute_kv.long_v from attribute_kv where attribute_kv.entity_id = %L and attribute_kv.attribute_key = %L', + tenant_id_record, 'TTL') INTO tenant_ttl; + if tenant_ttl IS NULL THEN + tenant_ttl := system_ttl; + END IF; + IF tenant_ttl > 0 THEN + tenant_ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - tenant_ttl::bigint * 1000)::bigint; + deleted_for_entities := delete_device_records_from_ts_kv(tenant_id_record, null_uuid, tenant_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for devices where tenant_id = %', deleted_for_entities, tenant_id_record; + deleted_for_entities := delete_asset_records_from_ts_kv(tenant_id_record, null_uuid, tenant_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for assets where tenant_id = %', deleted_for_entities, tenant_id_record; + END IF; + FOR customer_id_record IN + SELECT customer.id AS customer_id FROM customer WHERE customer.tenant_id = tenant_id_record + LOOP + EXECUTE format( + 'select attribute_kv.long_v from attribute_kv where attribute_kv.entity_id = %L and attribute_kv.attribute_key = %L', + customer_id_record, 'TTL') INTO customer_ttl; + IF customer_ttl IS NULL THEN + customer_ttl_ts := tenant_ttl_ts; + ELSE + IF customer_ttl > 0 THEN + customer_ttl_ts := + (EXTRACT(EPOCH FROM current_timestamp) * 1000 - + customer_ttl::bigint * 1000)::bigint; + END IF; + END IF; + IF customer_ttl_ts IS NOT NULL AND customer_ttl_ts > 0 THEN + deleted_for_entities := + delete_customer_records_from_ts_kv(tenant_id_record, customer_id_record, + customer_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for customer with id = % where tenant_id = %', deleted_for_entities, customer_id_record, tenant_id_record; + deleted_for_entities := + delete_device_records_from_ts_kv(tenant_id_record, customer_id_record, + customer_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for devices where tenant_id = % and customer_id = %', deleted_for_entities, tenant_id_record, customer_id_record; + deleted_for_entities := delete_asset_records_from_ts_kv(tenant_id_record, + customer_id_record, + customer_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for assets where tenant_id = % and customer_id = %', deleted_for_entities, tenant_id_record, customer_id_record; + END IF; + END LOOP; + FETCH tenant_cursor INTO tenant_id_record; + END LOOP; +END +$$; diff --git a/application/src/main/data/sql/schema-ts-psql.sql b/application/src/main/data/sql/schema-ts-psql.sql new file mode 100644 index 0000000000..c95ac4cf0f --- /dev/null +++ b/application/src/main/data/sql/schema-ts-psql.sql @@ -0,0 +1,339 @@ +-- +-- Copyright © 2016-2022 The Thingsboard Authors +-- +-- Licensed 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. +-- + +CREATE TABLE IF NOT EXISTS ts_kv +( + entity_id uuid NOT NULL, + key int NOT NULL, + ts bigint NOT NULL, + bool_v boolean, + str_v varchar(10000000), + long_v bigint, + dbl_v double precision, + json_v json, + CONSTRAINT ts_kv_pkey PRIMARY KEY (entity_id, key, ts) +) PARTITION BY RANGE (ts); + +CREATE TABLE IF NOT EXISTS ts_kv_dictionary +( + key varchar(255) NOT NULL, + key_id serial UNIQUE, + CONSTRAINT ts_key_id_pkey PRIMARY KEY (key) +); + +CREATE OR REPLACE PROCEDURE drop_partitions_by_max_ttl(IN partition_type varchar, IN system_ttl bigint, INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + max_tenant_ttl bigint; + max_customer_ttl bigint; + max_ttl bigint; + date timestamp; + partition_by_max_ttl_date varchar; + partition_by_max_ttl_month varchar; + partition_by_max_ttl_day varchar; + partition_by_max_ttl_year varchar; + partition varchar; + partition_year integer; + partition_month integer; + partition_day integer; + +BEGIN + SELECT max(attribute_kv.long_v) + FROM tenant + INNER JOIN attribute_kv ON tenant.id = attribute_kv.entity_id + WHERE attribute_kv.attribute_key = 'TTL' + into max_tenant_ttl; + SELECT max(attribute_kv.long_v) + FROM customer + INNER JOIN attribute_kv ON customer.id = attribute_kv.entity_id + WHERE attribute_kv.attribute_key = 'TTL' + into max_customer_ttl; + max_ttl := GREATEST(system_ttl, max_customer_ttl, max_tenant_ttl); + if max_ttl IS NOT NULL AND max_ttl > 0 THEN + date := to_timestamp(EXTRACT(EPOCH FROM current_timestamp) - max_ttl); + partition_by_max_ttl_date := get_partition_by_max_ttl_date(partition_type, date); + RAISE NOTICE 'Date by max ttl: %', date; + RAISE NOTICE 'Partition by max ttl: %', partition_by_max_ttl_date; + IF partition_by_max_ttl_date IS NOT NULL THEN + CASE + WHEN partition_type = 'DAYS' THEN + partition_by_max_ttl_year := SPLIT_PART(partition_by_max_ttl_date, '_', 3); + partition_by_max_ttl_month := SPLIT_PART(partition_by_max_ttl_date, '_', 4); + partition_by_max_ttl_day := SPLIT_PART(partition_by_max_ttl_date, '_', 5); + WHEN partition_type = 'MONTHS' THEN + partition_by_max_ttl_year := SPLIT_PART(partition_by_max_ttl_date, '_', 3); + partition_by_max_ttl_month := SPLIT_PART(partition_by_max_ttl_date, '_', 4); + ELSE + partition_by_max_ttl_year := SPLIT_PART(partition_by_max_ttl_date, '_', 3); + END CASE; + IF partition_by_max_ttl_year IS NULL THEN + RAISE NOTICE 'Failed to remove partitions by max ttl date due to partition_by_max_ttl_year is null!'; + ELSE + IF partition_type = 'YEARS' THEN + FOR partition IN SELECT tablename + FROM pg_tables + WHERE schemaname = 'public' + AND tablename like 'ts_kv_' || '%' + AND tablename != 'ts_kv_latest' + AND tablename != 'ts_kv_dictionary' + AND tablename != 'ts_kv_indefinite' + AND tablename != partition_by_max_ttl_date + LOOP + partition_year := SPLIT_PART(partition, '_', 3)::integer; + IF partition_year < partition_by_max_ttl_year::integer THEN + RAISE NOTICE 'Partition to delete by max ttl: %', partition; + EXECUTE format('DROP TABLE IF EXISTS %I', partition); + deleted := deleted + 1; + END IF; + END LOOP; + ELSE + IF partition_type = 'MONTHS' THEN + IF partition_by_max_ttl_month IS NULL THEN + RAISE NOTICE 'Failed to remove months partitions by max ttl date due to partition_by_max_ttl_month is null!'; + ELSE + FOR partition IN SELECT tablename + FROM pg_tables + WHERE schemaname = 'public' + AND tablename like 'ts_kv_' || '%' + AND tablename != 'ts_kv_latest' + AND tablename != 'ts_kv_dictionary' + AND tablename != 'ts_kv_indefinite' + AND tablename != partition_by_max_ttl_date + LOOP + partition_year := SPLIT_PART(partition, '_', 3)::integer; + IF partition_year > partition_by_max_ttl_year::integer THEN + RAISE NOTICE 'Skip iteration! Partition: % is valid!', partition; + CONTINUE; + ELSE + IF partition_year < partition_by_max_ttl_year::integer THEN + RAISE NOTICE 'Partition to delete by max ttl: %', partition; + EXECUTE format('DROP TABLE IF EXISTS %I', partition); + deleted := deleted + 1; + ELSE + partition_month := SPLIT_PART(partition, '_', 4)::integer; + IF partition_year = partition_by_max_ttl_year::integer THEN + IF partition_month >= partition_by_max_ttl_month::integer THEN + RAISE NOTICE 'Skip iteration! Partition: % is valid!', partition; + CONTINUE; + ELSE + RAISE NOTICE 'Partition to delete by max ttl: %', partition; + EXECUTE format('DROP TABLE IF EXISTS %I', partition); + deleted := deleted + 1; + END IF; + END IF; + END IF; + END IF; + END LOOP; + END IF; + ELSE + IF partition_type = 'DAYS' THEN + IF partition_by_max_ttl_month IS NULL THEN + RAISE NOTICE 'Failed to remove days partitions by max ttl date due to partition_by_max_ttl_month is null!'; + ELSE + IF partition_by_max_ttl_day IS NULL THEN + RAISE NOTICE 'Failed to remove days partitions by max ttl date due to partition_by_max_ttl_day is null!'; + ELSE + FOR partition IN SELECT tablename + FROM pg_tables + WHERE schemaname = 'public' + AND tablename like 'ts_kv_' || '%' + AND tablename != 'ts_kv_latest' + AND tablename != 'ts_kv_dictionary' + AND tablename != 'ts_kv_indefinite' + AND tablename != partition_by_max_ttl_date + LOOP + partition_year := SPLIT_PART(partition, '_', 3)::integer; + IF partition_year > partition_by_max_ttl_year::integer THEN + RAISE NOTICE 'Skip iteration! Partition: % is valid!', partition; + CONTINUE; + ELSE + IF partition_year < partition_by_max_ttl_year::integer THEN + RAISE NOTICE 'Partition to delete by max ttl: %', partition; + EXECUTE format('DROP TABLE IF EXISTS %I', partition); + deleted := deleted + 1; + ELSE + partition_month := SPLIT_PART(partition, '_', 4)::integer; + IF partition_month > partition_by_max_ttl_month::integer THEN + RAISE NOTICE 'Skip iteration! Partition: % is valid!', partition; + CONTINUE; + ELSE + IF partition_month < partition_by_max_ttl_month::integer THEN + RAISE NOTICE 'Partition to delete by max ttl: %', partition; + EXECUTE format('DROP TABLE IF EXISTS %I', partition); + deleted := deleted + 1; + ELSE + partition_day := SPLIT_PART(partition, '_', 5)::integer; + IF partition_day >= partition_by_max_ttl_day::integer THEN + RAISE NOTICE 'Skip iteration! Partition: % is valid!', partition; + CONTINUE; + ELSE + IF partition_day < partition_by_max_ttl_day::integer THEN + RAISE NOTICE 'Partition to delete by max ttl: %', partition; + EXECUTE format('DROP TABLE IF EXISTS %I', partition); + deleted := deleted + 1; + END IF; + END IF; + END IF; + END IF; + END IF; + END IF; + END LOOP; + END IF; + END IF; + END IF; + END IF; + END IF; + END IF; + END IF; + END IF; +END +$$; + +CREATE OR REPLACE FUNCTION get_partition_by_max_ttl_date(IN partition_type varchar, IN date timestamp, OUT partition varchar) AS +$$ +BEGIN + CASE + WHEN partition_type = 'DAYS' THEN + partition := 'ts_kv_' || to_char(date, 'yyyy') || '_' || to_char(date, 'MM') || '_' || to_char(date, 'dd'); + WHEN partition_type = 'MONTHS' THEN + partition := 'ts_kv_' || to_char(date, 'yyyy') || '_' || to_char(date, 'MM'); + WHEN partition_type = 'YEARS' THEN + partition := 'ts_kv_' || to_char(date, 'yyyy'); + ELSE + partition := NULL; + END CASE; + IF partition IS NOT NULL THEN + IF NOT EXISTS(SELECT + FROM pg_tables + WHERE schemaname = 'public' + AND tablename = partition) THEN + partition := NULL; + RAISE NOTICE 'Failed to found partition by ttl'; + END IF; + END IF; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION to_uuid(IN entity_id varchar, OUT uuid_id uuid) AS +$$ +BEGIN + uuid_id := substring(entity_id, 8, 8) || '-' || substring(entity_id, 4, 4) || '-1' || substring(entity_id, 1, 3) || + '-' || substring(entity_id, 16, 4) || '-' || substring(entity_id, 20, 12); +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION delete_device_records_from_ts_kv(tenant_id uuid, customer_id uuid, ttl bigint, + OUT deleted bigint) AS +$$ +BEGIN + EXECUTE format( + 'WITH deleted AS (DELETE FROM ts_kv WHERE entity_id IN (SELECT device.id as entity_id FROM device WHERE tenant_id = %L and customer_id = %L) AND ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', + tenant_id, customer_id, ttl) into deleted; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION delete_asset_records_from_ts_kv(tenant_id uuid, customer_id uuid, ttl bigint, + OUT deleted bigint) AS +$$ +BEGIN + EXECUTE format( + 'WITH deleted AS (DELETE FROM ts_kv WHERE entity_id IN (SELECT asset.id as entity_id FROM asset WHERE tenant_id = %L and customer_id = %L) AND ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', + tenant_id, customer_id, ttl) into deleted; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION delete_customer_records_from_ts_kv(tenant_id uuid, customer_id uuid, ttl bigint, + OUT deleted bigint) AS +$$ +BEGIN + EXECUTE format( + 'WITH deleted AS (DELETE FROM ts_kv WHERE entity_id IN (SELECT customer.id as entity_id FROM customer WHERE tenant_id = %L and id = %L) AND ts < %L::bigint RETURNING *) SELECT count(*) FROM deleted', + tenant_id, customer_id, ttl) into deleted; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE PROCEDURE cleanup_timeseries_by_ttl(IN null_uuid uuid, + IN system_ttl bigint, INOUT deleted bigint) + LANGUAGE plpgsql AS +$$ +DECLARE + tenant_cursor CURSOR FOR select tenant.id as tenant_id + from tenant; + tenant_id_record uuid; + customer_id_record uuid; + tenant_ttl bigint; + customer_ttl bigint; + deleted_for_entities bigint; + tenant_ttl_ts bigint; + customer_ttl_ts bigint; +BEGIN + OPEN tenant_cursor; + FETCH tenant_cursor INTO tenant_id_record; + WHILE FOUND + LOOP + EXECUTE format( + 'select attribute_kv.long_v from attribute_kv where attribute_kv.entity_id = %L and attribute_kv.attribute_key = %L', + tenant_id_record, 'TTL') INTO tenant_ttl; + if tenant_ttl IS NULL THEN + tenant_ttl := system_ttl; + END IF; + IF tenant_ttl > 0 THEN + tenant_ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - tenant_ttl::bigint * 1000)::bigint; + deleted_for_entities := delete_device_records_from_ts_kv(tenant_id_record, null_uuid, tenant_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for devices where tenant_id = %', deleted_for_entities, tenant_id_record; + deleted_for_entities := delete_asset_records_from_ts_kv(tenant_id_record, null_uuid, tenant_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for assets where tenant_id = %', deleted_for_entities, tenant_id_record; + END IF; + FOR customer_id_record IN + SELECT customer.id AS customer_id FROM customer WHERE customer.tenant_id = tenant_id_record + LOOP + EXECUTE format( + 'select attribute_kv.long_v from attribute_kv where attribute_kv.entity_id = %L and attribute_kv.attribute_key = %L', + customer_id_record, 'TTL') INTO customer_ttl; + IF customer_ttl IS NULL THEN + customer_ttl_ts := tenant_ttl_ts; + ELSE + IF customer_ttl > 0 THEN + customer_ttl_ts := + (EXTRACT(EPOCH FROM current_timestamp) * 1000 - + customer_ttl::bigint * 1000)::bigint; + END IF; + END IF; + IF customer_ttl_ts IS NOT NULL AND customer_ttl_ts > 0 THEN + deleted_for_entities := + delete_customer_records_from_ts_kv(tenant_id_record, customer_id_record, + customer_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for customer with id = % where tenant_id = %', deleted_for_entities, customer_id_record, tenant_id_record; + deleted_for_entities := + delete_device_records_from_ts_kv(tenant_id_record, customer_id_record, + customer_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for devices where tenant_id = % and customer_id = %', deleted_for_entities, tenant_id_record, customer_id_record; + deleted_for_entities := delete_asset_records_from_ts_kv(tenant_id_record, + customer_id_record, + customer_ttl_ts); + deleted := deleted + deleted_for_entities; + RAISE NOTICE '% telemetry removed for assets where tenant_id = % and customer_id = %', deleted_for_entities, tenant_id_record, customer_id_record; + END IF; + END LOOP; + FETCH tenant_cursor INTO tenant_id_record; + END LOOP; +END +$$; diff --git a/application/src/main/java/org/thingsboard/server/controller/OtaPackageController.java b/application/src/main/java/org/thingsboard/server/controller/OtaPackageController.java index fa60ea3033..719d56510d 100644 --- a/application/src/main/java/org/thingsboard/server/controller/OtaPackageController.java +++ b/application/src/main/java/org/thingsboard/server/controller/OtaPackageController.java @@ -46,6 +46,7 @@ import org.thingsboard.server.common.data.ota.ChecksumAlgorithm; import org.thingsboard.server.common.data.ota.OtaPackageType; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.dao.ota.util.ChecksumUtil; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.cache.ota.service.FileCacheService; import org.thingsboard.server.service.security.permission.Operation; @@ -95,12 +96,12 @@ public class OtaPackageController extends BaseController { checkParameter(OTA_PACKAGE_ID, strOtaPackageId); try { OtaPackageId otaPackageId = new OtaPackageId(toUUID(strOtaPackageId)); -// OtaPackage otaPackage = checkOtaPackageId(otaPackageId, Operation.READ); - OtaPackage otaPackage = otaPackageService.findOtaPackageById(new TenantId(UUID.fromString("b04858a0-b646-11ec-8d4d-f5c10326c05e")), otaPackageId); + OtaPackage otaPackage = checkOtaPackageId(otaPackageId, Operation.READ); if (otaPackage.hasUrl()) { return ResponseEntity.badRequest().build(); } - InputStreamResource resource = fileCacheService.getOtaResourceById(new TenantId(UUID.fromString("b04858a0-b646-11ec-8d4d-f5c10326c05e")), otaPackageId); + InputStream dataStream = otaPackageService.getOtaDataStream(getTenantId(), otaPackageId); + InputStreamResource resource = new InputStreamResource(dataStream); return ResponseEntity.ok() .header(HttpHeaders.CONTENT_DISPOSITION, "attachment;filename=" + otaPackage.getFileName()) .header("x-filename", otaPackage.getFileName()) @@ -197,7 +198,7 @@ public class OtaPackageController extends BaseController { ChecksumAlgorithm checksumAlgorithm = ChecksumAlgorithm.valueOf(checksumAlgorithmStr.toUpperCase()); if (StringUtils.isEmpty(checksum)) { - checksum = otaPackageService.generateChecksum(checksumAlgorithm, file.getInputStream()); + checksum = ChecksumUtil.generateChecksum(checksumAlgorithm, file.getInputStream()); } OtaPackageInfo savedOtaPackage = saveOtaPackageWithData(otaPackageId, file, checksum, checksumAlgorithm); logEntityAction(savedOtaPackage.getId(), savedOtaPackage, null, ActionType.UPDATED, null); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 03c3bbd9f9..4c181c0218 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -505,8 +505,8 @@ spring.resources.chain: content: enabled: "true" -spring.servlet.multipart.max-file-size: "50MB" -spring.servlet.multipart.max-request-size: "50MB" +spring.servlet.multipart.max-file-size: "2000MB" +spring.servlet.multipart.max-request-size: "2000MB" spring.jpa.properties.hibernate.jdbc.lob.non_contextual_creation: "true" spring.jpa.properties.hibernate.order_by.default_null_ordering: "${SPRING_JPA_PROPERTIES_HIBERNATE_ORDER_BY_DEFAULT_NULL_ORDERING:last}" @@ -526,9 +526,9 @@ spring: database-platform: "${SPRING_JPA_DATABASE_PLATFORM:org.hibernate.dialect.PostgreSQLDialect}" datasource: driverClassName: "${SPRING_DRIVER_CLASS_NAME:org.postgresql.Driver}" - url: "${SPRING_DATASOURCE_URL:jdbc:postgresql://localhost:5432/thingsboard}" - username: "${SPRING_DATASOURCE_USERNAME:postgres}" - password: "${SPRING_DATASOURCE_PASSWORD:postgres}" + url: "${SPRING_DATASOURCE_URL:jdbc:postgresql://localhost:5432/board}" + username: "${SPRING_DATASOURCE_USERNAME:myuser}" + password: "${SPRING_DATASOURCE_PASSWORD:mypass}" hikari: maximumPoolSize: "${SPRING_DATASOURCE_MAXIMUM_POOL_SIZE:16}" diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java index 7ea7b18f43..5c4e5c2e66 100644 --- a/common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java +++ b/common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java @@ -1,12 +1,12 @@ /** * Copyright © 2016-2022 The Thingsboard Authors - * + *

* Licensed 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 - * + *

+ * 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. diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/ota/service/FileCacheService.java b/common/cache/src/main/java/org/thingsboard/server/cache/ota/service/FileCacheService.java index dd9f1eb049..485360b2d0 100644 --- a/common/cache/src/main/java/org/thingsboard/server/cache/ota/service/FileCacheService.java +++ b/common/cache/src/main/java/org/thingsboard/server/cache/ota/service/FileCacheService.java @@ -1,12 +1,12 @@ /** * Copyright © 2016-2022 The Thingsboard Authors - * + *

* Licensed 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 - * + *

+ * 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. @@ -18,12 +18,9 @@ package org.thingsboard.server.cache.ota.service; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.io.FileUtils; -import org.springframework.core.io.InputStreamResource; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; -import org.thingsboard.server.common.data.OtaPackage; import org.thingsboard.server.common.data.id.OtaPackageId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.dao.ota.OtaPackageService; import javax.annotation.PostConstruct; import java.io.*; @@ -31,6 +28,7 @@ import java.sql.Blob; import java.sql.SQLException; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.concurrent.*; import java.util.stream.Collectors; @@ -38,12 +36,40 @@ import java.util.stream.Collectors; @RequiredArgsConstructor @Component public class FileCacheService { - private final static String PATH = "files/%s"; - private final static String FILE_NAME_TEMPLATE = "%s.tmp"; - private final static long TEMPORARY_FILE_INACTIVITY_TIME = 300; + @Value("${java.io.tmpdir}/ota/") + private String PATH; + private final static String FILE_PATH_TEMPLATE = "%s"; + private final static long TEMPORARY_FILE_INACTIVITY_TIME = 300;//todo increase time private final ConcurrentMap files = new ConcurrentHashMap<>(); private final ConcurrentMap lastActivityTimes = new ConcurrentHashMap<>(); - private final OtaPackageService otaPackageService; + + @PostConstruct + private void createScheduledTaskToClear() { + createTempDirectoryIfNotExist(); + cleanDirectoryWithTemporaryFiles(); + ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); + scheduler.scheduleAtFixedRate(this::deleteUnusedTemporaryFiles, 3, 3, TimeUnit.MINUTES); //todo increase time + } + + private void createTempDirectoryIfNotExist() { + File directory = new File(PATH); + if(!directory.exists()){ + directory.mkdir(); + } + } + + private void cleanDirectoryWithTemporaryFiles() { + File directory = new File(PATH); + if (directory.isDirectory()) { + try { + FileUtils.cleanDirectory(directory); + log.info("Directory with temporary files cleaned"); + } catch (IOException e) { + log.error("Failed to clean directory with temporary files", e); + throw new RuntimeException(e); + } + } + } private void deleteUnusedTemporaryFiles() { long currentTime = System.currentTimeMillis(); @@ -57,67 +83,47 @@ public class FileCacheService { deleteFile(otaId.getId().toString()); lastActivityTimes.remove(otaId); }); - }catch (Exception e){ + } catch (Exception e) { log.error(",,", e); } log.info("Deleted {} unused temporary files", toBeDeleted.size()); } private void deleteFile(String otaId) { - if (exist(otaId)) { - String fileName = String.format(FILE_NAME_TEMPLATE, otaId); - File file = new File(String.format(PATH, fileName)); - file.delete(); - } + String fileName = PATH + otaId; + File file = new File(String.format(FILE_PATH_TEMPLATE, fileName)); + deleteFile(file); } - private void cleanDirectoryWithTemporaryFiles() { - File directory = new File(String.format(PATH, "")); - if (directory.isDirectory()) { - try { - FileUtils.cleanDirectory(directory); - log.info("Directory with temporary files cleaned"); - } catch (IOException e) { - log.error("Failed to clean directory with temporary files", e); - throw new RuntimeException(e); + public void deleteFile(File file) { + try { + if (file.exists()) { + FileUtils.delete(file); + log.info("System file {} was deleted", file.getName()); } + } catch (IOException e) { + log.error("Failed to delete file {}", file.getName(), e); } } - - @PostConstruct - private void createScheduledTaskToClear() { - cleanDirectoryWithTemporaryFiles(); - ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); - scheduler.scheduleAtFixedRate(this::deleteUnusedTemporaryFiles, 3, 3, TimeUnit.MINUTES); - } - - public InputStreamResource getOtaResourceById(TenantId tenantId, OtaPackageId otaPackageId) { - try { - String fileName = String.format(FILE_NAME_TEMPLATE, otaPackageId.getId().toString()); - if (exist(fileName)) { - lastActivityTimes.put(otaPackageId, System.currentTimeMillis()); - return new InputStreamResource(new FileInputStream(String.format(PATH, fileName))); - } - InputStreamResource resource = load(tenantId, otaPackageId); + public InputStream getOtaDataStream(OtaPackageId otaPackageId) throws FileNotFoundException { + String fileName = PATH + otaPackageId.getId(); + if (exist(fileName)) { lastActivityTimes.put(otaPackageId, System.currentTimeMillis()); - files.remove(otaPackageId); - return resource; - } catch (FileNotFoundException e) { - log.error("Failed to find resource for otaPackage {}", otaPackageId, e); - throw new RuntimeException(e); + return new FileInputStream(String.format(FILE_PATH_TEMPLATE, fileName)); } + throw new FileNotFoundException(); } - private InputStreamResource load(TenantId tenantId, OtaPackageId otaPackageId) throws FileNotFoundException { - CountDownLatch latch = files.computeIfAbsent(otaPackageId, ota -> processFileSaving(tenantId, ota)); - String fileName = String.format(FILE_NAME_TEMPLATE, otaPackageId.getId().toString()); + private InputStream load(OtaPackageId otaPackageId, Blob data) throws FileNotFoundException { + CountDownLatch latch = files.computeIfAbsent(otaPackageId, ota -> processFileSaving(ota, data)); + String fileName = PATH + otaPackageId.getId(); if (latch.getCount() == 0) { - return new InputStreamResource(new FileInputStream(String.format(PATH, fileName))); + return new FileInputStream(String.format(FILE_PATH_TEMPLATE, fileName)); } else { try { latch.await(); - return new InputStreamResource(new FileInputStream(String.format(PATH, fileName))); + return new FileInputStream(String.format(FILE_PATH_TEMPLATE, fileName)); } catch (InterruptedException e) { log.error("Failed to save file {} to file system", fileName, e); throw new RuntimeException(e); @@ -125,18 +131,11 @@ public class FileCacheService { } } - private CountDownLatch processFileSaving(TenantId tenantId, OtaPackageId otaPackageId) { + private CountDownLatch processFileSaving(OtaPackageId otaPackageId, Blob data) { CountDownLatch latch = new CountDownLatch(1); - OtaPackage otaPackage = otaPackageService.findOtaPackageById(tenantId, otaPackageId); - if (otaPackage == null) { - latch.countDown(); - log.error("Can't find otaPackage to download file {}", otaPackage); - throw new RuntimeException("No such OtaPackageId"); - } try { - Blob data = otaPackage.getData(); InputStream binaryStream = data.getBinaryStream(); - String fileName = String.format(FILE_NAME_TEMPLATE, otaPackageId.getId().toString()); + String fileName =PATH + otaPackageId.getId(); saveAsSystemFile(fileName, binaryStream); latch.countDown(); return latch; @@ -148,7 +147,7 @@ public class FileCacheService { private void saveAsSystemFile(String fileName, InputStream inputStream) { try { - File file = new File(String.format(PATH, fileName)); + File file = new File(String.format(FILE_PATH_TEMPLATE, fileName)); FileUtils.copyInputStreamToFile(inputStream, file); } catch (IOException e) { log.error("Failed to copy stream to system file {}", fileName, e); @@ -157,7 +156,28 @@ public class FileCacheService { } private boolean exist(String name) { - File file = new File(String.format(PATH, name)); + File file = new File(String.format(FILE_PATH_TEMPLATE, name)); return file.exists(); } + + public InputStream loadOtaData(OtaPackageId otaPackageId, Blob data) { + try { + return load(otaPackageId, data); + } catch (FileNotFoundException e) { + log.error("Failed to upload temporary file {} to system", otaPackageId.getId(), e); + throw new RuntimeException(e); + } + } + + public File saveDataTemporaryFile(InputStream inputStream) { + File path = new File(PATH); + try { + File tempFile = File.createTempFile(UUID.randomUUID().toString(), ".tmp", path); + FileUtils.copyInputStreamToFile(inputStream, tempFile); + return tempFile; + } catch (IOException e) { + log.error("Failed to create temp file", e); + throw new RuntimeException("Failed to create temp file for input stream"); + } + } } \ No newline at end of file diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/ota/OtaPackageService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/ota/OtaPackageService.java index 8e8038cf32..8ce5e64066 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/ota/OtaPackageService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/ota/OtaPackageService.java @@ -1,12 +1,12 @@ /** * Copyright © 2016-2022 The Thingsboard Authors - * + *

* Licensed 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 - * + *

+ * 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. @@ -52,5 +52,6 @@ public interface OtaPackageService { void deleteOtaPackagesByTenantId(TenantId tenantId); long sumDataSizeByTenantId(TenantId tenantId); - String generateChecksum(ChecksumAlgorithm checksumAlgorithm, InputStream fileData); + + InputStream getOtaDataStream(TenantId tenantId, OtaPackageId otaPackageId); } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/ota/ChecksumAlgorithm.java b/common/data/src/main/java/org/thingsboard/server/common/data/ota/ChecksumAlgorithm.java index 6c764ccd3f..73477b5cc6 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/ota/ChecksumAlgorithm.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/ota/ChecksumAlgorithm.java @@ -1,12 +1,12 @@ /** * Copyright © 2016-2022 The Thingsboard Authors - * + *

* Licensed 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 - * + *

+ * 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. @@ -15,12 +15,21 @@ */ package org.thingsboard.server.common.data.ota; +import lombok.Getter; + public enum ChecksumAlgorithm { - MD5, - SHA256, - SHA384, - SHA512, - CRC32, - MURMUR3_32, - MURMUR3_128 + MD5("MD5"), + SHA256("SHA-256"), + SHA384("SHA-384"), + SHA512("SHA-512"), + CRC32("CRC32"), + MURMUR3_32("MURMUR3-32"), + MURMUR3_128("MURMUR3_128"); + + @Getter + final String name; + + ChecksumAlgorithm(String name) { + this.name = name; + } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java b/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java index 6a1c14a639..ffa6906cd0 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java @@ -28,6 +28,7 @@ import org.springframework.cache.CacheManager; import org.springframework.cache.annotation.Cacheable; import org.springframework.stereotype.Service; import org.thingsboard.server.cache.ota.OtaPackageDataCache; +import org.thingsboard.server.cache.ota.service.FileCacheService; import org.thingsboard.server.common.data.OtaPackage; import org.thingsboard.server.common.data.OtaPackageInfo; import org.thingsboard.server.common.data.StringUtils; @@ -42,6 +43,7 @@ import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.service.DataValidator; import org.thingsboard.server.dao.service.PaginatedRemover; +import javax.transaction.Transactional; import java.io.*; import java.nio.ByteBuffer; import java.security.MessageDigest; @@ -62,15 +64,13 @@ import static org.thingsboard.server.dao.service.Validator.validatePageLink; public class BaseOtaPackageService implements OtaPackageService { public static final String INCORRECT_OTA_PACKAGE_ID = "Incorrect otaPackageId "; public static final String INCORRECT_TENANT_ID = "Incorrect tenantId "; - - private static final int ONE_MEGA_BYTE = 1_000_000; - private final OtaPackageDao otaPackageDao; private final OtaPackageInfoDao otaPackageInfoDao; private final CacheManager cacheManager; private final OtaPackageDataCache otaPackageDataCache; private final DataValidator otaPackageInfoValidator; private final DataValidator otaPackageValidator; + private final FileCacheService fileCacheService; @Override public OtaPackageInfo saveOtaPackageInfo(OtaPackageInfo otaPackageInfo, boolean isUrl) { @@ -101,12 +101,12 @@ public class BaseOtaPackageService implements OtaPackageService { public OtaPackage saveOtaPackage(OtaPackage otaPackage) { log.trace("Executing saveOtaPackage [{}]", otaPackage); try { - File tempFile = saveDataToTemporaryFile(otaPackage.getData().getBinaryStream()); + File tempFile= fileCacheService.saveDataTemporaryFile(otaPackage.getData().getBinaryStream()); otaPackage.setData(BlobProxy.generateProxy(new FileInputStream(tempFile), otaPackage.getDataSize())); try { otaPackageValidator.validate(otaPackage, OtaPackageInfo::getTenantId); }catch (RuntimeException e){ - deleteTempFile(tempFile); + fileCacheService.deleteFile(tempFile); throw e; } otaPackage.setData(BlobProxy.generateProxy(new FileInputStream(tempFile), otaPackage.getDataSize())); @@ -117,7 +117,7 @@ public class BaseOtaPackageService implements OtaPackageService { otaPackageDataCache.evict(otaPackageId.toString()); } OtaPackage save = otaPackageDao.save(otaPackage.getTenantId(), otaPackage); - deleteTempFile(tempFile); + fileCacheService.deleteFile(tempFile); return save; } catch (FileNotFoundException | SQLException e){ log.error("Failed to validate ota package {}",otaPackage.getId(), e); @@ -131,31 +131,6 @@ public class BaseOtaPackageService implements OtaPackageService { } } } - - - public void deleteTempFile(File file) { - try { - if(file.exists()) { - FileUtils.delete(file); - log.info("System file {} was deleted", file.getName()); - } - } catch (IOException e) { - log.error("Failed to delete file {}", file.getName(), e); - } - } - - public File saveDataToTemporaryFile(InputStream inputStream){ - File path = new File("files/"); - try { - File tempFile = File.createTempFile(UUID.randomUUID().toString(), ".tmp", path); - FileUtils.copyInputStreamToFile(inputStream, tempFile); - return tempFile; - }catch (IOException e){ - log.error("Failed to create temp file", e); - throw new RuntimeException("Failed to create temp file for input stream"); - } - } - @Override public String generateChecksum(ChecksumAlgorithm checksumAlgorithm, ByteBuffer data) { if (data == null || !data.hasArray() || data.array().length == 0) { @@ -290,33 +265,17 @@ public class BaseOtaPackageService implements OtaPackageService { return Collections.singletonList(otaPackageId); } - public String generateChecksum(ChecksumAlgorithm checksumAlgorithm, InputStream fileData) { + @Transactional + public InputStream getOtaDataStream(TenantId tenantId, OtaPackageId otaPackageId){ try { - MessageDigest md = MessageDigest.getInstance(checksumAlgorithm.name()); - return checksum(fileData, md); - } catch (NoSuchAlgorithmException e) { - log.error("No such checksum algorithm {}", checksumAlgorithm, e); - throw new RuntimeException(e); - } catch (IOException e) { - log.error("Failed to calculate checksum", e); - throw new RuntimeException(e); - } catch (Exception e) { - log.error("ff", e); - throw new RuntimeException(e); - } - } - - private String checksum(InputStream inputStream, MessageDigest md) throws IOException { - byte[] buffer = new byte[ONE_MEGA_BYTE]; - int count = 0; - while ((count = inputStream.read(buffer)) != -1) { - md.update(buffer, 0, count); - } - StringBuilder result = new StringBuilder(); - for (byte b : md.digest()) { - result.append(String.format("%02x", b)); + return fileCacheService.getOtaDataStream(otaPackageId); + }catch (FileNotFoundException e){ + OtaPackage otaPackage = findOtaPackageById(tenantId, otaPackageId); + if (otaPackage == null) { + log.error("Can't find otaPackage to download file {}", otaPackage); + throw new RuntimeException("No such OtaPackageId"); + } + return fileCacheService.loadOtaData(otaPackageId,otaPackage.getData()); } - return result.toString(); } - } diff --git a/dao/src/main/java/org/thingsboard/server/dao/ota/util/ChecksumUtil.java b/dao/src/main/java/org/thingsboard/server/dao/ota/util/ChecksumUtil.java new file mode 100644 index 0000000000..9d3c7469e2 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/ota/util/ChecksumUtil.java @@ -0,0 +1,79 @@ +package org.thingsboard.server.dao.ota.util; + +import com.datastax.oss.driver.shaded.guava.common.io.ByteStreams; +import com.google.common.hash.Funnels; +import com.google.common.hash.Hasher; +import com.google.common.hash.Hashing; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.ota.ChecksumAlgorithm; + +import java.io.IOException; +import java.io.InputStream; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.Base64; +import java.util.zip.CRC32; + +@Slf4j +public class ChecksumUtil { + private static final int ONE_MEGA_BYTE = 1_000_000; + + public static String generateChecksum(ChecksumAlgorithm checksumAlgorithm, InputStream fileData) { + try { + switch (checksumAlgorithm) { + case CRC32: + return checksumCRC32(fileData); + case MURMUR3_128: + return checksumMurmur3_128(fileData); + case MURMUR3_32: + return checksumMurmur3_32(fileData); + default: + MessageDigest md = MessageDigest.getInstance(checksumAlgorithm.getName()); + return checksum(fileData, md); + } + } catch (NoSuchAlgorithmException e) { + log.error("No such checksum algorithm {}", checksumAlgorithm, e); + throw new RuntimeException(e); + } catch (Exception e) { + log.error("Failed to calculate checksum", e); + throw new RuntimeException(e); + } + } + + private static String checksum(InputStream inputStream, MessageDigest md) throws IOException { + byte[] buffer = new byte[ONE_MEGA_BYTE]; + int count = 0; + while ((count = inputStream.read(buffer)) != -1) { + md.update(buffer, 0, count); + } + StringBuilder result = new StringBuilder(); + for (byte b : md.digest()) { + result.append(String.format("%02x", b)); + } + return result.toString(); + } + + private static String checksumCRC32(InputStream inputStream) throws IOException { + CRC32 crc = new CRC32(); + byte[] buffer = new byte[ONE_MEGA_BYTE]; + int count = 0; + while ((count = inputStream.read(buffer)) != -1) { + crc.update(buffer, 0, count); + } + return Long.toHexString(crc.getValue()); + } + + private static String checksumMurmur3_32(InputStream stream) throws IOException { + Hasher hasher = Hashing.murmur3_32().newHasher(); + ByteStreams.copy(stream, Funnels.asOutputStream(hasher)); + String hash = new String(hasher.hash().asBytes()); + return new String(Base64.getEncoder().encode(hash.getBytes())); + } + + private static String checksumMurmur3_128(InputStream stream) throws IOException { + Hasher hasher = Hashing.murmur3_128().newHasher(); + ByteStreams.copy(stream, Funnels.asOutputStream(hasher)); + String hash = new String(hasher.hash().asBytes()); + return new String(Base64.getEncoder().encode(hash.getBytes())); + } +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.java b/dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.java index 449d005ec1..e55b5e333a 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileCon import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.ota.OtaPackageDao; import org.thingsboard.server.dao.ota.OtaPackageService; +import org.thingsboard.server.dao.ota.util.ChecksumUtil; import org.thingsboard.server.dao.tenant.TbTenantProfileCache; @@ -41,10 +42,6 @@ public class OtaPackageDataValidator extends BaseOtaPackageDataValidatorChecksumUtil.generateChecksum(MURMUR3_32, inputStream)); + } + + @Test + void testMurmur3_128Checksum(){ + assertDoesNotThrow(()->ChecksumUtil.generateChecksum(MURMUR3_128, inputStream)); + } + +} \ No newline at end of file diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java index c1ecafda6e..03783cf833 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java @@ -19,6 +19,7 @@ import com.datastax.oss.driver.api.core.uuid.Uuids; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.commons.lang3.RandomStringUtils; +import org.hibernate.engine.jdbc.BlobProxy; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -268,7 +269,7 @@ public abstract class AbstractServiceTest { firmware.setContentType("text/plain"); firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256); firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a"); - firmware.setData(ByteBuffer.wrap(new byte[]{1})); + firmware.setData(BlobProxy.generateProxy(new byte[]{1})); firmware.setDataSize(1L); return firmware; } diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceProfileServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceProfileServiceTest.java index 4e0b0e7dfc..400286c13e 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceProfileServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceProfileServiceTest.java @@ -19,6 +19,7 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; +import org.hibernate.engine.jdbc.BlobProxy; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -110,7 +111,7 @@ public abstract class BaseDeviceProfileServiceTest extends AbstractServiceTest { firmware.setContentType("text/plain"); firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256); firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a"); - firmware.setData(ByteBuffer.wrap(new byte[]{1})); + firmware.setData(BlobProxy.generateProxy(new byte[]{1})); firmware.setDataSize(1L); OtaPackage savedFirmware = otaPackageService.saveOtaPackage(firmware); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceServiceTest.java index b4c0ff77dd..283ac8fd0e 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceServiceTest.java @@ -17,6 +17,7 @@ package org.thingsboard.server.dao.service; import com.datastax.oss.driver.api.core.uuid.Uuids; import org.apache.commons.lang3.RandomStringUtils; +import org.hibernate.engine.jdbc.BlobProxy; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -200,7 +201,7 @@ public abstract class BaseDeviceServiceTest extends AbstractServiceTest { firmware.setContentType("text/plain"); firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256); firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a"); - firmware.setData(ByteBuffer.wrap(new byte[]{1})); + firmware.setData(BlobProxy.generateProxy(new byte[]{1})); firmware.setDataSize(1L); OtaPackage savedFirmware = otaPackageService.saveOtaPackage(firmware); @@ -235,7 +236,7 @@ public abstract class BaseDeviceServiceTest extends AbstractServiceTest { firmware.setContentType("text/plain"); firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256); firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a"); - firmware.setData(ByteBuffer.wrap(new byte[]{1})); + firmware.setData(BlobProxy.generateProxy(new byte[]{1})); firmware.setDataSize(1L); OtaPackage savedFirmware = otaPackageService.saveOtaPackage(firmware); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseOtaPackageServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseOtaPackageServiceTest.java index 7dedf14bab..006392bb45 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseOtaPackageServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseOtaPackageServiceTest.java @@ -17,6 +17,7 @@ package org.thingsboard.server.dao.service; import com.datastax.oss.driver.api.core.uuid.Uuids; import org.apache.commons.lang3.RandomStringUtils; +import org.hibernate.engine.jdbc.BlobProxy; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -40,6 +41,7 @@ import org.thingsboard.server.dao.exception.DataValidationException; import javax.validation.ValidationException; import java.nio.ByteBuffer; +import java.sql.Blob; import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -56,7 +58,7 @@ public abstract class BaseOtaPackageServiceTest extends AbstractServiceTest { private static final ChecksumAlgorithm CHECKSUM_ALGORITHM = ChecksumAlgorithm.SHA256; private static final String CHECKSUM = "4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a"; private static final long DATA_SIZE = 1L; - private static final ByteBuffer DATA = ByteBuffer.wrap(new byte[]{(int) DATA_SIZE}); + private static final Blob DATA = BlobProxy.generateProxy(new byte[]{(int) DATA_SIZE}); private static final String URL = "http://firmware.test.org"; private final IdComparator idComparator = new IdComparator<>();