tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

http sending + checksum improved

pull/7062/head
AnastasiiaKomar 4 years ago
parent
b047ccfd7e
commit
82225075ec
19 changed files with 1571 additions and 160 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 38
      application/src/main/data/sql/schema-entities-idx-psql-addon.sql
  2. 53
      application/src/main/data/sql/schema-entities-idx.sql
  3. 695
      application/src/main/data/sql/schema-entities.sql
  4. 157
      application/src/main/data/sql/schema-timescale.sql
  5. 339
      application/src/main/data/sql/schema-ts-psql.sql
  6. 9
      application/src/main/java/org/thingsboard/server/controller/OtaPackageController.java
  7. 10
      application/src/main/resources/thingsboard.yml
  8. 8
      common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java
  9. 148
      common/cache/src/main/java/org/thingsboard/server/cache/ota/service/FileCacheService.java
  10. 11
      common/dao-api/src/main/java/org/thingsboard/server/dao/ota/OtaPackageService.java
  11. 31
      common/data/src/main/java/org/thingsboard/server/common/data/ota/ChecksumAlgorithm.java
  12. 73
      dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java
  13. 79
      dao/src/main/java/org/thingsboard/server/dao/ota/util/ChecksumUtil.java
  14. 7
      dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.java
  15. 58
      dao/src/test/java/org/thingsboard/server/dao/ota/util/ChecksumUtilTest.java
  16. 3
      dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java
  17. 3
      dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceProfileServiceTest.java
  18. 5
      dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceServiceTest.java
  19. 4
      dao/src/test/java/org/thingsboard/server/dao/service/BaseOtaPackageServiceTest.java

38
application/src/main/data/sql/schema-entities-idx-psql-addon.sql
View File

@ -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';

53
application/src/main/data/sql/schema-entities-idx.sql
View File

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

695
application/src/main/data/sql/schema-entities.sql
View File

@ -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
$$;

157
application/src/main/data/sql/schema-timescale.sql
View File

@ -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
$$;

339
application/src/main/data/sql/schema-ts-psql.sql
View File

@ -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
$$;

9
application/src/main/java/org/thingsboard/server/controller/OtaPackageController.java
View File

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

10
application/src/main/resources/thingsboard.yml
View File

@ -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}"

8
common/cache/src/main/java/org/thingsboard/server/cache/ota/CaffeineOtaPackageCache.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* <p>
* 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
*
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.

148
common/cache/src/main/java/org/thingsboard/server/cache/ota/service/FileCacheService.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* <p>
* 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
*
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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<OtaPackageId, CountDownLatch> files = new ConcurrentHashMap<>();
private final ConcurrentMap<OtaPackageId, Long> 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");
}
}
}

11
common/dao-api/src/main/java/org/thingsboard/server/dao/ota/OtaPackageService.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* <p>
* 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
*
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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);
}

31
common/data/src/main/java/org/thingsboard/server/common/data/ota/ChecksumAlgorithm.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2022 The Thingsboard Authors
*
* <p>
* 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
*
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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;
}
}

73
dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java
View File

@ -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<OtaPackageInfo> otaPackageInfoValidator;
private final DataValidator<OtaPackage> 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();
}
}

79
dao/src/main/java/org/thingsboard/server/dao/ota/util/ChecksumUtil.java
View File

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

7
dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.java
View File

@ -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 BaseOtaPackageDataValidator<OtaPack
@Autowired
private OtaPackageDao otaPackageDao;
@Autowired
@Lazy
private OtaPackageService otaPackageService;
@Autowired
@Lazy
private TbTenantProfileCache tenantProfileCache;
@ -77,7 +74,7 @@ public class OtaPackageDataValidator extends BaseOtaPackageDataValidator<OtaPack
throw new DataValidationException("OtaPackage checksum should be specified!");
}
try {
String currentChecksum = otaPackageService.generateChecksum(otaPackage.getChecksumAlgorithm(), otaPackage.getData().getBinaryStream());
String currentChecksum = ChecksumUtil.generateChecksum(otaPackage.getChecksumAlgorithm(), otaPackage.getData().getBinaryStream());
if (!currentChecksum.equals(otaPackage.getChecksum())) {
throw new DataValidationException("Wrong otaPackage file!");
}

58
dao/src/test/java/org/thingsboard/server/dao/ota/util/ChecksumUtilTest.java
View File

@ -0,0 +1,58 @@
package org.thingsboard.server.dao.ota.util;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.*;
import static org.thingsboard.server.common.data.ota.ChecksumAlgorithm.*;
import java.io.ByteArrayInputStream;
class ChecksumUtilTest {
ByteArrayInputStream inputStream;
@BeforeEach
void setUp() {
inputStream =
new ByteArrayInputStream("ggvgavsckvawyefgkashcbweyfguas;cnewgfayegfdhvzhxeyrgfzkdfgybycefygkzbvdrgzksdfbfvxjchvksydfvdkzsf".getBytes());
}
@Test
void testSha256Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(SHA256, inputStream);
assertEquals("8098be8d5e22198e69316b1eef0b2863d75973b708f2ec4849132122b45fea71", generateChecksum);
}
@Test
void testMd5Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(MD5, inputStream);
assertEquals("64c03752176a3a66e627db31ad4d3a9d", generateChecksum);
}
@Test
void testSha384Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(SHA384, inputStream);
assertEquals("f7a88d89699d5fb9d4fb3b84b606bb1cac6641ed04e51f598fb7729e2142d5b0cec0c4019801d239749ff6ea0a872604", generateChecksum);
}
@Test
void testSha512Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(SHA512, inputStream);
assertEquals("5709738c21dfc8ea5b80a291a0e11c4c38bc8393d01b075e712877bf4b6329f822ab1a6d8eecd0d578e7ae9a51fff852f3c9849d23f5109a8068f92a858a88f0",
generateChecksum);
}
@Test
void testCrc32Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(CRC32, inputStream);
assertEquals("666124cf", generateChecksum);
}
@Test
void testMurmur3_32Checksum(){
assertDoesNotThrow(()->ChecksumUtil.generateChecksum(MURMUR3_32, inputStream));
}
@Test
void testMurmur3_128Checksum(){
assertDoesNotThrow(()->ChecksumUtil.generateChecksum(MURMUR3_128, inputStream));
}
}

3
dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java
View File

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

3
dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceProfileServiceTest.java
View File

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

5
dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceServiceTest.java
View File

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

4
dao/src/test/java/org/thingsboard/server/dao/service/BaseOtaPackageServiceTest.java
View File

@ -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<OtaPackageInfo> idComparator = new IdComparator<>();

Write Preview
Loading…
Cancel
Save