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

First CR

pull/7062/head
AnastasiiaKomar 4 years ago
parent
49f700eba1
commit
2e2174248e
28 changed files with 306 additions and 1745 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. 62
      application/src/main/java/org/thingsboard/server/controller/OtaPackageController.java
  7. 27
      application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
  8. 8
      application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
  9. 13
      application/src/main/resources/thingsboard.yml
  10. 15
      common/cache/pom.xml
  11. 100
      common/cache/src/main/java/org/thingsboard/server/cache/ota/files/BaseFileCacheService.java
  12. 14
      common/cache/src/main/java/org/thingsboard/server/cache/ota/files/FileCacheService.java
  13. 49
      common/cache/src/main/java/org/thingsboard/server/cache/ota/files/TemporaryFileCleaner.java
  14. 137
      common/cache/src/main/java/org/thingsboard/server/cache/ota/service/BaseFileCacheService.java
  15. 17
      common/cache/src/main/java/org/thingsboard/server/cache/ota/service/FileCacheService.java
  16. 47
      common/cache/src/test/java/org/thingsboard/server/cache/ota/service/BaseFileCacheServiceTest.java
  17. 7
      common/dao-api/src/main/java/org/thingsboard/server/dao/ota/OtaPackageService.java
  18. 18
      common/data/src/main/java/org/thingsboard/server/common/data/OtaPackage.java
  19. 1
      common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java
  20. 49
      dao/src/main/java/org/thingsboard/server/dao/model/sql/OtaPackageEntity.java
  21. 112
      dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.java
  22. 8
      dao/src/main/java/org/thingsboard/server/dao/ota/util/ChecksumUtil.java
  23. 24
      dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.java
  24. 36
      dao/src/test/java/org/thingsboard/server/dao/ota/util/ChecksumUtilTest.java
  25. 3
      dao/src/test/java/org/thingsboard/server/dao/service/AbstractServiceTest.java
  26. 3
      dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceProfileServiceTest.java
  27. 5
      dao/src/test/java/org/thingsboard/server/dao/service/BaseDeviceServiceTest.java
  28. 14
      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

@ -1,38 +0,0 @@
--
-- 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

@ -1,53 +0,0 @@
--
-- 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

@ -1,695 +0,0 @@
--
-- 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

@ -1,157 +0,0 @@
--
-- 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

@ -1,339 +0,0 @@
--
-- 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
$$;

62
application/src/main/java/org/thingsboard/server/controller/OtaPackageController.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.
@ -20,17 +20,11 @@ import io.swagger.annotations.ApiParam;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.hibernate.engine.jdbc.BlobProxy;
import org.springframework.core.io.FileSystemResource;
import org.springframework.http.*;
import org.springframework.http.HttpHeaders;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.OtaPackage;
@ -50,27 +44,9 @@ import org.thingsboard.server.service.security.permission.Operation;
import org.thingsboard.server.service.security.permission.Resource;
import javax.transaction.Transactional;
import java.io.InputStream;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.sql.Blob;
import java.io.IOException;
import static org.thingsboard.server.controller.ControllerConstants.DEVICE_PROFILE_ID_PARAM_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.OTA_PACKAGE_CHECKSUM_ALGORITHM_ALLOWABLE_VALUES;
import static org.thingsboard.server.controller.ControllerConstants.OTA_PACKAGE_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.OTA_PACKAGE_ID_PARAM_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.OTA_PACKAGE_INFO_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.OTA_PACKAGE_SORT_PROPERTY_ALLOWABLE_VALUES;
import static org.thingsboard.server.controller.ControllerConstants.OTA_PACKAGE_TEXT_SEARCH_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.PAGE_DATA_PARAMETERS;
import static org.thingsboard.server.controller.ControllerConstants.PAGE_NUMBER_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.PAGE_SIZE_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_ALLOWABLE_VALUES;
import static org.thingsboard.server.controller.ControllerConstants.SORT_ORDER_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.SORT_PROPERTY_DESCRIPTION;
import static org.thingsboard.server.controller.ControllerConstants.TENANT_AUTHORITY_PARAGRAPH;
import static org.thingsboard.server.controller.ControllerConstants.TENANT_OR_CUSTOMER_AUTHORITY_PARAGRAPH;
import static org.thingsboard.server.controller.ControllerConstants.UUID_WIKI_LINK;
import static org.thingsboard.server.controller.ControllerConstants.*;
@Slf4j
@RestController
@ -89,7 +65,7 @@ public class OtaPackageController extends BaseController {
@ResponseBody
@Transactional
public ResponseEntity<FileSystemResource> downloadOtaPackage(@ApiParam(value = OTA_PACKAGE_ID_PARAM_DESCRIPTION)
@PathVariable(OTA_PACKAGE_ID) String strOtaPackageId) throws ThingsboardException {
@PathVariable(OTA_PACKAGE_ID) String strOtaPackageId) throws ThingsboardException {
checkParameter(OTA_PACKAGE_ID, strOtaPackageId);
try {
OtaPackageId otaPackageId = new OtaPackageId(toUUID(strOtaPackageId));
@ -205,7 +181,7 @@ public class OtaPackageController extends BaseController {
}
}
private OtaPackage saveOtaPackageWithData(OtaPackageId otaPackageId, MultipartFile file, String checksum, ChecksumAlgorithm checksumAlgorithm) throws ThingsboardException {
private OtaPackage saveOtaPackageWithData(OtaPackageId otaPackageId, MultipartFile file, String checksum, ChecksumAlgorithm checksumAlgorithm) throws ThingsboardException, IOException {
OtaPackageInfo info = checkOtaPackageInfoId(otaPackageId, Operation.READ);
OtaPackage otaPackage = new OtaPackage(otaPackageId);
otaPackage.setCreatedTime(info.getCreatedTime());
@ -218,17 +194,11 @@ public class OtaPackageController extends BaseController {
otaPackage.setAdditionalInfo(info.getAdditionalInfo());
otaPackage.setChecksumAlgorithm(checksumAlgorithm);
otaPackage.setChecksum(checksum);
try(InputStream inputStream = file.getInputStream()) {
otaPackage.setFileName(file.getOriginalFilename());
otaPackage.setContentType(file.getContentType());
otaPackage.setDataSize(file.getSize());
Blob blob = BlobProxy.generateProxy(inputStream, file.getSize());
otaPackage.setData(blob);
return otaPackageService.saveOtaPackage(otaPackage);
} catch (Exception e){
log.error("Failed to save ota package {} with data {}",otaPackageId, file.getName(), e);
return null;
}
otaPackage.setFileName(file.getOriginalFilename());
otaPackage.setContentType(file.getContentType());
otaPackage.setDataSize(file.getSize());
otaPackage.setData(file.getInputStream());
return otaPackageService.saveOtaPackage(otaPackage);
}
@ApiOperation(value = "Get OTA Package Infos (getOtaPackages)",

27
application/src/main/java/org/thingsboard/server/service/action/EntityActionService.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.
@ -22,9 +22,9 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.*;
import org.thingsboard.server.common.data.audit.ActionType;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
@ -37,7 +37,6 @@ import org.thingsboard.server.common.msg.TbMsgDataType;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.dao.audit.AuditLogService;
import org.thingsboard.server.queue.util.TbCoreComponent;
import org.thingsboard.server.cluster.TbClusterService;
import java.util.List;
import java.util.Map;
@ -155,15 +154,9 @@ public class EntityActionService {
}
ObjectNode entityNode;
if (entity != null) {
if (entity instanceof OtaPackage) {
OtaPackage otaPackage = (OtaPackage) entity;
otaPackage.setData(null);
entityNode = json.valueToTree(otaPackage); //todo
} else {
entityNode = json.valueToTree(entity);
if (entityId.getEntityType() == EntityType.DASHBOARD) {
entityNode.put("configuration", "");
}
entityNode = json.valueToTree(entity);
if (entityId.getEntityType() == EntityType.DASHBOARD) {
entityNode.put("configuration", "");
}
} else {
entityNode = json.createObjectNode();
@ -214,8 +207,8 @@ public class EntityActionService {
}
}
public <E extends HasName, I extends EntityId> void logEntityAction(User user, I entityId, E entity, CustomerId customerId,
ActionType actionType, Exception e, Object... additionalInfo) {
public <E extends HasName, I extends EntityId> void logEntityAction(User user, I entityId, E entity, CustomerId customerId,
ActionType actionType, Exception e, Object... additionalInfo) {
if (customerId == null || customerId.isNullUid()) {
customerId = user.getCustomerId();
}

8
application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
View File

@ -99,6 +99,7 @@ import org.thingsboard.server.service.executors.DbCallbackExecutorService;
import org.thingsboard.server.service.profile.TbDeviceProfileCache;
import org.thingsboard.server.service.resource.TbResourceService;
import java.io.IOException;
import java.sql.SQLException;
import java.util.Optional;
import java.util.UUID;
@ -573,7 +574,6 @@ public class DefaultTransportApiService implements TransportApiService {
DeviceProfile deviceProfile = deviceProfileCache.find(device.getDeviceProfileId());
otaPackageId = OtaPackageUtil.getOtaPackageId(deviceProfile, otaPackageType);
}
TransportProtos.GetOtaPackageResponseMsg.Builder builder = TransportProtos.GetOtaPackageResponseMsg.newBuilder();
if (otaPackageId == null) {
@ -598,9 +598,9 @@ public class DefaultTransportApiService implements TransportApiService {
if (!otaPackageDataCache.has(otaPackageId.toString())) {
OtaPackage otaPackage = otaPackageService.findOtaPackageById(tenantId, otaPackageId);
try {
otaPackageDataCache.put(otaPackageId.toString(), otaPackage.getData().getBytes(100L, 0));//todo
} catch (SQLException e) {
log.error("ERROR", e); // todo
otaPackageDataCache.put(otaPackageId.toString(), otaPackage.getData().readAllBytes());
} catch (IOException e) {
log.error("Failed to cache ota package with id {}",otaPackage.getId(), e);
}
}
}

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

@ -505,8 +505,8 @@ spring.resources.chain:
content:
enabled: "true"
spring.servlet.multipart.max-file-size: "2000MB"
spring.servlet.multipart.max-request-size: "2000MB"
spring.servlet.multipart.max-file-size: "50MB"
spring.servlet.multipart.max-request-size: "50MB"
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/board}"
username: "${SPRING_DATASOURCE_USERNAME:myuser}"
password: "${SPRING_DATASOURCE_PASSWORD:mypass}"
url: "${SPRING_DATASOURCE_URL:jdbc:postgresql://localhost:5432/thingsboard}"
username: "${SPRING_DATASOURCE_USERNAME:postgres}"
password: "${SPRING_DATASOURCE_PASSWORD:postgres}"
hikari:
maximumPoolSize: "${SPRING_DATASOURCE_MAXIMUM_POOL_SIZE:16}"
@ -1114,3 +1114,6 @@ management:
exposure:
# Expose metrics endpoint (use value 'prometheus' to enable prometheus metrics).
include: '${METRICS_ENDPOINTS_EXPOSE:info}'
files:
temporary_files_directory: '${TEMPORARY_FILES_DIRECTORY: ${java.io.tmpdir}}'

15
common/cache/pom.xml
View File

@ -40,14 +40,15 @@
<groupId>org.thingsboard.common</groupId>
<artifactId>data</artifactId>
</dependency>
<dependency>
<groupId>org.thingsboard.common</groupId>
<artifactId>dao-api</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
<version>1.3.2</version>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
@ -95,12 +96,6 @@
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.hibernate</groupId>
<artifactId>hibernate-core</artifactId>
<version>5.4.32.Final</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>

100
common/cache/src/main/java/org/thingsboard/server/cache/ota/files/BaseFileCacheService.java
View File

@ -0,0 +1,100 @@
/**
* 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
* <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.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.cache.ota.files;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.id.OtaPackageId;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@Slf4j
@RequiredArgsConstructor
@Component
public class BaseFileCacheService implements FileCacheService {
@Value("${files.temporary_files_directory}/ota/")
private String PATH;
private final static String FILE_NAME_TEMPLATE = "%s.tmp";
private final TemporaryFileCleaner fileCleaner;
private final ConcurrentMap<OtaPackageId, Boolean> files = new ConcurrentHashMap<>();
@Override
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");
}
}
@Override
public Optional<File> getOtaDataFile(OtaPackageId otaPackageId) {
String fileName = PATH + String.format(FILE_NAME_TEMPLATE,otaPackageId.getId().toString());
if (exist(fileName)) {
fileCleaner.updateFileUsageStatus(otaPackageId);
return Optional.of(new File(fileName));
}
return Optional.empty();
}
@Override
public File loadToFile(OtaPackageId otaPackageId, InputStream data) {
if (otaPackageId == null || data == null) {
log.error("Received null variables: {}", otaPackageId == null ? "otaPackageId" : "data");
throw new RuntimeException("Input values can not be null");
}
files.computeIfAbsent(otaPackageId, ota -> processFileSaving(ota, data));
String fileName = PATH + String.format(FILE_NAME_TEMPLATE,otaPackageId.getId().toString());
return new File(fileName);
}
private Boolean processFileSaving(OtaPackageId otaPackageId, InputStream data) {
String fileName = PATH + String.format(FILE_NAME_TEMPLATE,otaPackageId.getId().toString());
saveAsSystemFile(fileName, data);
fileCleaner.updateFileUsageStatus(otaPackageId);
return true;
}
private void saveAsSystemFile(String fileName, InputStream inputStream) {
try {
File file = new File(fileName);
FileUtils.copyInputStreamToFile(inputStream, file);
} catch (IOException e) {
log.error("Failed to copy stream to system file {}", fileName, e);
throw new RuntimeException("Failed to save file");
}
}
private boolean exist(String name) {
File file = new File(name);
return file.exists();
}
}

14
common/cache/src/main/java/org/thingsboard/server/cache/ota/files/FileCacheService.java
View File

@ -0,0 +1,14 @@
package org.thingsboard.server.cache.ota.files;
import org.thingsboard.server.common.data.id.OtaPackageId;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.util.Optional;
public interface FileCacheService {
File loadToFile(OtaPackageId otaPackageId, InputStream data);
File saveDataTemporaryFile(InputStream inputStream);
Optional<File> getOtaDataFile(OtaPackageId otaPackageId) throws FileNotFoundException;
}

49
common/cache/src/main/java/org/thingsboard/server/cache/ota/service/TemporaryFileCleaner.java → common/cache/src/main/java/org/thingsboard/server/cache/ota/files/TemporaryFileCleaner.java
View File

@ -1,38 +1,44 @@
package org.thingsboard.server.cache.ota.service;
package org.thingsboard.server.cache.ota.files;
import lombok.extern.apachecommons.CommonsLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.id.OtaPackageId;
import javax.annotation.PostConstruct;
import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
@Slf4j
@Component
public class TemporaryFileCleaner {
@Value("${java.io.tmpdir}/ota/")
@Value("${files.temporary_files_directory}/ota/")
private String PATH;
private final static long TEMPORARY_FILE_INACTIVITY_TIME = 900_000;
private final ConcurrentMap<OtaPackageId, Long> lastActivityTimes = new ConcurrentHashMap<>();
public void updateFileUsageStatus(OtaPackageId otaPackageId){
public void updateFileUsageStatus(OtaPackageId otaPackageId) {
lastActivityTimes.put(otaPackageId, System.currentTimeMillis());
}
@PostConstruct
@EventListener(ApplicationReadyEvent.class)
public void createScheduledTaskToClear() {
createTempDirectoryIfNotExist();
cleanDirectoryWithTemporaryFiles();
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(this::deleteUnusedTemporaryFiles, 10, 10, TimeUnit.MINUTES);
}
private void createTempDirectoryIfNotExist() {
@ -45,16 +51,21 @@ public class TemporaryFileCleaner {
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);
}
File[] files = directory.listFiles();
if (files == null) return;
Arrays.stream(files).forEach(
file -> {
try {
FileUtils.delete(file);
} catch (Exception e) {
log.error("Failed to delete file {}", file.getName(), e);
}
}
);
}
}
@Scheduled(fixedDelay = 60000)
private void deleteUnusedTemporaryFiles() {
long currentTime = System.currentTimeMillis();
List<OtaPackageId> toBeDeleted = lastActivityTimes.entrySet()
@ -73,14 +84,16 @@ public class TemporaryFileCleaner {
log.info("Deleted {} unused temporary files", toBeDeleted.size());
}
private void deleteFile(String otaId) {
private synchronized void deleteFile(String otaId) {
String fileName = PATH + otaId;
File file = new File(fileName);
try {
try (FileChannel channel = FileChannel.open(Path.of(URI.create(file.getPath())), StandardOpenOption.APPEND)) {
FileLock lock = channel.lock();
if (file.exists()) {
FileUtils.delete(file);
log.info("System file {} was deleted", file.getName());
}
lock.release();
} catch (IOException e) {
log.error("Failed to delete file {}", file.getName(), e);
}

137
common/cache/src/main/java/org/thingsboard/server/cache/ota/service/BaseFileCacheService.java
View File

@ -1,137 +0,0 @@
/**
* 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
* <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.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.cache.ota.service;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.thingsboard.server.common.data.id.OtaPackageId;
import java.io.*;
import java.sql.Blob;
import java.sql.SQLException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
@Slf4j
@RequiredArgsConstructor
@Component
public class BaseFileCacheService implements FileCacheService {
@Value("${java.io.tmpdir}/ota/")
private String PATH;
private final TemporaryFileCleaner fileCleaner;
private final ConcurrentMap<OtaPackageId, CountDownLatch> files = new ConcurrentHashMap<>();
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");
}
}
public File getOtaDataFile(OtaPackageId otaPackageId) throws FileNotFoundException {
String fileName = PATH + otaPackageId.getId();
if (exist(fileName)) {
fileCleaner.updateFileUsageStatus(otaPackageId);
return new File(fileName);
}
throw new FileNotFoundException();
}
public File loadToFile(OtaPackageId otaPackageId, Blob data) {
if (otaPackageId == null || data == null) {
log.error("Received null variables: {}", otaPackageId == null ? "otaPackageId" : "data");
throw new NullPointerException();
}
CountDownLatch latch = files.computeIfAbsent(otaPackageId, ota -> processFileSaving(ota, data));
String fileName = PATH + otaPackageId.getId();
if (latch.getCount() == 0) {
return new File(fileName);
} else {
try {
latch.await();
return new File(fileName);
} catch (InterruptedException e) {
log.error("Failed to save file {} to file system", fileName, 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);
}
}
public InputStream getOtaDataStream(OtaPackageId otaPackageId) throws FileNotFoundException {
return new FileInputStream(getOtaDataFile(otaPackageId));
}
public InputStream loadOtaData(OtaPackageId otaPackageId, Blob data) {
try {
return new FileInputStream(loadToFile(otaPackageId, data));
} catch (FileNotFoundException e) {
log.error("Failed to upload temporary file {} to system", otaPackageId.getId(), e);
throw new RuntimeException(e);
}
}
private CountDownLatch processFileSaving(OtaPackageId otaPackageId, Blob data) {
CountDownLatch latch = new CountDownLatch(1);
try {
InputStream binaryStream = data.getBinaryStream();
String fileName = PATH + otaPackageId.getId();
saveAsSystemFile(fileName, binaryStream);
latch.countDown();
return latch;
} catch (SQLException e) {
log.error("Failed to get binary data from otaPackage {}", otaPackageId, e);
throw new RuntimeException("Failed to get binary data from database");
}
}
private void saveAsSystemFile(String fileName, InputStream inputStream) {
try {
File file = new File(fileName);
FileUtils.copyInputStreamToFile(inputStream, file);
} catch (IOException e) {
log.error("Failed to copy stream to system file {}", fileName, e);
throw new RuntimeException("Failed to save file");
}
}
private boolean exist(String name) {
File file = new File(name);
return file.exists();
}
}

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

@ -1,17 +0,0 @@
package org.thingsboard.server.cache.ota.service;
import org.thingsboard.server.common.data.id.OtaPackageId;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.sql.Blob;
public interface FileCacheService {
void deleteFile(File file);
InputStream getOtaDataStream(OtaPackageId otaPackageId) throws FileNotFoundException ;
InputStream loadOtaData(OtaPackageId otaPackageId, Blob data);
File loadToFile(OtaPackageId otaPackageId, Blob data);
File saveDataTemporaryFile(InputStream inputStream);
File getOtaDataFile(OtaPackageId otaPackageId) throws FileNotFoundException;
}

47
common/cache/src/test/java/org/thingsboard/server/cache/ota/service/BaseFileCacheServiceTest.java
View File

@ -1,19 +1,21 @@
package org.thingsboard.server.cache.ota.service;
import lombok.SneakyThrows;
import org.hibernate.engine.jdbc.BlobProxy;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import org.thingsboard.server.cache.ota.files.BaseFileCacheService;
import org.thingsboard.server.cache.ota.files.TemporaryFileCleaner;
import org.thingsboard.server.common.data.id.OtaPackageId;
import java.io.*;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.security.MessageDigest;
import java.sql.Blob;
import java.util.Objects;
import java.util.UUID;
import static org.junit.jupiter.api.Assertions.*;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
class BaseFileCacheServiceTest {
@ -21,30 +23,18 @@ class BaseFileCacheServiceTest {
private final static String FILE_FILLING = "Hello, testing environment";
private static final int ONE_MEGA_BYTE = 1_000_000;
private final static OtaPackageId OTA_PACKAGE_ID = new OtaPackageId(UUID.randomUUID());
private final static Blob DATA = BlobProxy.generateProxy(FILE_FILLING.getBytes());
File file;
private final static InputStream DATA = new ByteArrayInputStream(FILE_FILLING.getBytes());
BaseFileCacheService baseFileCacheService = new BaseFileCacheService(new TemporaryFileCleaner());
@BeforeEach
void setUp() {
file = Mockito.mock(File.class);
// fileCacheService = Mockito.mock(FileCacheService.class);
}
@Test
void testNotThrowExceptionsWhenDeleteFile() {
Mockito.when(file.exists()).thenReturn(true);
assertDoesNotThrow(() -> baseFileCacheService.deleteFile(new File("/download/alarm_rule.js")));
}
@Test
void testDataSavingWithNullBlob() {
assertThrows(NullPointerException.class, () -> baseFileCacheService.loadToFile(OTA_PACKAGE_ID, null));
void testDataSavingWithNullInputStream() {
assertThrows(RuntimeException.class, () -> baseFileCacheService.loadToFile(OTA_PACKAGE_ID, null));
}
@Test
void testDataSavingWithNullOtaPackageId() {
assertThrows(NullPointerException.class, () -> baseFileCacheService.loadToFile(null, DATA));
assertThrows(RuntimeException.class, () -> baseFileCacheService.loadToFile(null, DATA));
}
@Test
@ -71,19 +61,6 @@ class BaseFileCacheServiceTest {
assertEquals(sha256, calculateChecksumSHA256(new FileInputStream(file)));
}
@Test
@SneakyThrows
void testNotThrowExceptionWhenCreatingTemporaryFiles() {
BaseFileCacheService mock = Mockito.mock(BaseFileCacheService.class);
assertDoesNotThrow(() -> mock.saveDataTemporaryFile(DATA.getBinaryStream()));
}
@Test
@SneakyThrows
void testGettingStreamWhenFileNotExist(){
assertThrows(FileNotFoundException.class, ()->baseFileCacheService.getOtaDataStream(OTA_PACKAGE_ID));
}
@SneakyThrows
String calculateChecksumSHA256(InputStream stream) {
MessageDigest md = MessageDigest.getInstance("SHA-256");

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

@ -21,14 +21,12 @@ import org.thingsboard.server.common.data.OtaPackageInfo;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.id.TenantId;
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 java.io.File;
import java.io.InputStream;
import java.nio.ByteBuffer;
public interface OtaPackageService {
@ -36,8 +34,6 @@ public interface OtaPackageService {
OtaPackage saveOtaPackage(OtaPackage otaPackage);
String generateChecksum(ChecksumAlgorithm checksumAlgorithm, ByteBuffer data);
OtaPackage findOtaPackageById(TenantId tenantId, OtaPackageId otaPackageId);
OtaPackageInfo findOtaPackageInfoById(TenantId tenantId, OtaPackageId otaPackageId);
@ -54,6 +50,5 @@ public interface OtaPackageService {
long sumDataSizeByTenantId(TenantId tenantId);
InputStream getOtaDataStream(TenantId tenantId, OtaPackageId otaPackageId);
File getOtaDataFile (TenantId tenantId, OtaPackageId otaPackageId);
File getOtaDataFile(TenantId tenantId, OtaPackageId otaPackageId);
}

18
common/data/src/main/java/org/thingsboard/server/common/data/OtaPackage.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,24 +15,28 @@
*/
package org.thingsboard.server.common.data;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.thingsboard.server.common.data.id.OtaPackageId;
import java.nio.ByteBuffer;
import java.sql.Blob;
import java.io.InputStream;
@ApiModel
@Data
@EqualsAndHashCode(callSuper = true)
@JsonIgnoreProperties(value = { "data" })
public class OtaPackage extends OtaPackageInfo {
private static final long serialVersionUID = 3091601761339422546L;
@ApiModelProperty(position = 16, value = "OTA Package data.", readOnly = true)
private transient Blob data;
private transient InputStream data;
public OtaPackage() {
super();

1
common/util/src/main/java/org/thingsboard/common/util/JacksonUtil.java
View File

@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.IOException;

49
dao/src/main/java/org/thingsboard/server/dao/model/sql/OtaPackageEntity.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,44 +18,32 @@ package org.thingsboard.server.dao.model.sql;
import com.fasterxml.jackson.databind.JsonNode;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.SneakyThrows;
import org.apache.commons.io.FileUtils;
import org.aspectj.util.FileUtil;
import org.hibernate.annotations.Type;
import org.hibernate.annotations.TypeDef;
import org.hibernate.engine.jdbc.BlobProxy;
import org.postgresql.largeobject.BlobInputStream;
import org.springframework.beans.factory.annotation.Value;
import org.thingsboard.server.common.data.OtaPackage;
import org.thingsboard.server.common.data.ota.ChecksumAlgorithm;
import org.thingsboard.server.common.data.ota.OtaPackageType;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.ota.ChecksumAlgorithm;
import org.thingsboard.server.common.data.ota.OtaPackageType;
import org.thingsboard.server.dao.model.BaseSqlEntity;
import org.thingsboard.server.dao.model.ModelConstants;
import org.thingsboard.server.dao.model.SearchTextEntity;
import org.thingsboard.server.dao.util.mapping.JsonStringType;
import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.EnumType;
import javax.persistence.Enumerated;
import javax.persistence.Lob;
import javax.persistence.Table;
import java.nio.ByteBuffer;
import javax.persistence.*;
import java.io.File;
import java.io.FileInputStream;
import java.sql.Blob;
import java.util.UUID;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_CHECKSUM_ALGORITHM_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_CHECKSUM_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_CONTENT_TYPE_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_DATA_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_DATA_SIZE_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_DEVICE_PROFILE_ID_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_FILE_NAME_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_TABLE_NAME;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_TAG_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_TENANT_ID_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_TILE_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_TYPE_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_URL_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.OTA_PACKAGE_VERSION_COLUMN;
import static org.thingsboard.server.dao.model.ModelConstants.SEARCH_TEXT_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.*;
@Data
@EqualsAndHashCode(callSuper = true)
@ -133,7 +121,7 @@ public class OtaPackageEntity extends BaseSqlEntity<OtaPackage> implements Searc
this.contentType = otaPackage.getContentType();
this.checksumAlgorithm = otaPackage.getChecksumAlgorithm();
this.checksum = otaPackage.getChecksum();
this.data = otaPackage.getData();
this.data = BlobProxy.generateProxy(otaPackage.getData(), otaPackage.getDataSize());
this.dataSize = otaPackage.getDataSize();
this.additionalInfo = otaPackage.getAdditionalInfo();
}
@ -149,6 +137,7 @@ public class OtaPackageEntity extends BaseSqlEntity<OtaPackage> implements Searc
}
@Override
@SneakyThrows
public OtaPackage toData() {
OtaPackage otaPackage = new OtaPackage(new OtaPackageId(id));
otaPackage.setCreatedTime(createdTime);
@ -167,7 +156,7 @@ public class OtaPackageEntity extends BaseSqlEntity<OtaPackage> implements Searc
otaPackage.setChecksum(checksum);
otaPackage.setDataSize(dataSize);
if (data != null) {
otaPackage.setData(data);
otaPackage.setData(data.getBinaryStream());
otaPackage.setHasData(true);
}
otaPackage.setAdditionalInfo(additionalInfo);

112
dao/src/main/java/org/thingsboard/server/dao/ota/BaseOtaPackageService.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,26 +15,22 @@
*/
package org.thingsboard.server.dao.ota;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.hibernate.engine.jdbc.BlobProxy;
import org.hibernate.exception.ConstraintViolationException;
import org.springframework.cache.Cache;
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.BaseFileCacheService;
import org.thingsboard.server.cache.ota.files.BaseFileCacheService;
import org.thingsboard.server.common.data.OtaPackage;
import org.thingsboard.server.common.data.OtaPackageInfo;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.id.TenantId;
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;
@ -43,9 +39,10 @@ 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.sql.SQLException;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
@ -71,7 +68,7 @@ public class BaseOtaPackageService implements OtaPackageService {
@Override
public OtaPackageInfo saveOtaPackageInfo(OtaPackageInfo otaPackageInfo, boolean isUrl) {
log.trace("Executing saveOtaPackageInfo [{}]", otaPackageInfo);
if(isUrl && (StringUtils.isEmpty(otaPackageInfo.getUrl()) || otaPackageInfo.getUrl().trim().length() == 0)) {
if (isUrl && (StringUtils.isEmpty(otaPackageInfo.getUrl()) || otaPackageInfo.getUrl().trim().length() == 0)) {
throw new DataValidationException("Ota package URL should be specified!");
}
otaPackageInfoValidator.validate(otaPackageInfo, OtaPackageInfo::getTenantId);
@ -94,29 +91,26 @@ public class BaseOtaPackageService implements OtaPackageService {
}
@Override
@Transactional
public OtaPackage saveOtaPackage(OtaPackage otaPackage) {
log.trace("Executing saveOtaPackage [{}]", otaPackage);
try {
File tempFile= baseFileCacheService.saveDataTemporaryFile(otaPackage.getData().getBinaryStream());
otaPackage.setData(BlobProxy.generateProxy(new FileInputStream(tempFile), otaPackage.getDataSize()));
try {
otaPackageValidator.validate(otaPackage, OtaPackageInfo::getTenantId);
}catch (RuntimeException e){
baseFileCacheService.deleteFile(tempFile);
throw e;
}
otaPackage.setData(BlobProxy.generateProxy(new FileInputStream(tempFile), otaPackage.getDataSize()));
File tempFile = baseFileCacheService.saveDataTemporaryFile(otaPackage.getData());
otaPackage.setData(new FileInputStream(tempFile));
otaPackageValidator.validate(otaPackage, OtaPackageInfo::getTenantId);
BufferedInputStream stream = new BufferedInputStream(new FileInputStream(tempFile));
stream.mark(0);
otaPackage.setData(stream);
OtaPackageId otaPackageId = otaPackage.getId();
if (otaPackageId != null) {
Cache cache = cacheManager.getCache(OTA_PACKAGE_CACHE);
cache.evict(toOtaPackageInfoKey(otaPackageId));
otaPackageDataCache.evict(otaPackageId.toString());
}
OtaPackage save = otaPackageDao.save(otaPackage.getTenantId(), otaPackage);
baseFileCacheService.deleteFile(tempFile);
return save;
} catch (FileNotFoundException | SQLException e){
log.error("Failed to validate ota package {}",otaPackage.getId(), e);
OtaPackage savedOtaPackage = otaPackageDao.save(otaPackage.getTenantId(), otaPackage);
return savedOtaPackage;
} catch (FileNotFoundException e) {
log.error("Failed to save data of ota package {} to file", otaPackage.getId(), e);
throw new RuntimeException(e);
} catch (Exception t) {
ConstraintViolationException e = extractConstraintViolationException(t).orElse(null);
@ -127,36 +121,6 @@ public class BaseOtaPackageService implements OtaPackageService {
}
}
}
@Override
public String generateChecksum(ChecksumAlgorithm checksumAlgorithm, ByteBuffer data) {
if (data == null || !data.hasArray() || data.array().length == 0) {
throw new DataValidationException("OtaPackage data should be specified!");
}
return getHashFunction(checksumAlgorithm).hashBytes(data.array()).toString();
}
@SuppressWarnings("deprecation")
private HashFunction getHashFunction(ChecksumAlgorithm checksumAlgorithm) {
switch (checksumAlgorithm) {
case MD5:
return Hashing.md5();
case SHA256:
return Hashing.sha256();
case SHA384:
return Hashing.sha384();
case SHA512:
return Hashing.sha512();
case CRC32:
return Hashing.crc32();
case MURMUR3_32:
return Hashing.murmur3_32();
case MURMUR3_128:
return Hashing.murmur3_128();
default:
throw new DataValidationException("Unknown checksum algorithm!");
}
}
@Override
public OtaPackage findOtaPackageById(TenantId tenantId, OtaPackageId otaPackageId) {
@ -261,31 +225,19 @@ public class BaseOtaPackageService implements OtaPackageService {
return Collections.singletonList(otaPackageId);
}
@Override
@Transactional
public InputStream getOtaDataStream(TenantId tenantId, OtaPackageId otaPackageId){
try {
return baseFileCacheService.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 baseFileCacheService.loadOtaData(otaPackageId,otaPackage.getData());
public File getOtaDataFile(TenantId tenantId, OtaPackageId otaPackageId) {
validateId(otaPackageId, INCORRECT_OTA_PACKAGE_ID + otaPackageId);
Optional<File> otaDataFile = baseFileCacheService.getOtaDataFile(otaPackageId);
if (otaDataFile.isPresent()) {
return otaDataFile.get();
}
}
@Transactional
public File getOtaDataFile(TenantId tenantId, OtaPackageId otaPackageId){
try {
return baseFileCacheService.getOtaDataFile(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 baseFileCacheService.loadToFile(otaPackageId,otaPackage.getData());
OtaPackage otaPackage = findOtaPackageById(tenantId, otaPackageId);
if (otaPackage == null) {
log.error("Can't find otaPackage to download file {}", otaPackageId);
throw new RuntimeException("No such OtaPackageId");
}
return baseFileCacheService.loadToFile(otaPackageId, otaPackage.getData());
}
}

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

@ -65,15 +65,13 @@ public class ChecksumUtil {
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()));
com.google.common.io.ByteStreams.copy(stream, Funnels.asOutputStream(hasher));
return hasher.hash().toString();
}
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()));
return hasher.hash().toString();
}
}

24
dao/src/main/java/org/thingsboard/server/dao/service/validator/OtaPackageDataValidator.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.
@ -25,14 +25,9 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
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;
import java.sql.SQLException;
import static org.thingsboard.server.common.data.EntityType.OTA_PACKAGE;
@Component
@ -73,14 +68,9 @@ public class OtaPackageDataValidator extends BaseOtaPackageDataValidator<OtaPack
if (StringUtils.isEmpty(otaPackage.getChecksum())) {
throw new DataValidationException("OtaPackage checksum should be specified!");
}
try {
String currentChecksum = ChecksumUtil.generateChecksum(otaPackage.getChecksumAlgorithm(), otaPackage.getData().getBinaryStream());
if (!currentChecksum.equals(otaPackage.getChecksum())) {
throw new DataValidationException("Wrong otaPackage file!");
}
}catch (SQLException e){
log.error("Failed to check ota package data {}",otaPackage.getId(), e);
throw new RuntimeException("otaPackage file can't be validated");
String currentChecksum = ChecksumUtil.generateChecksum(otaPackage.getChecksumAlgorithm(), otaPackage.getData());
if (!currentChecksum.equals(otaPackage.getChecksum())) {
throw new DataValidationException("Wrong otaPackage file!");
}
} else {
if (otaPackage.getData() != null) {

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

@ -2,57 +2,67 @@ 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;
import java.util.Arrays;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.thingsboard.server.common.data.ota.ChecksumAlgorithm.*;
class ChecksumUtilTest {
private static final int SIZE = 1_050_000;
ByteArrayInputStream inputStream;
@BeforeEach
void setUp() {
inputStream =
new ByteArrayInputStream("ggvgavsckvawyefgkashcbweyfguas;cnewgfayegfdhvzhxeyrgfzkdfgybycefygkzbvdrgzksdfbfvxjchvksydfvdkzsf".getBytes());
char[] chars = new char[SIZE];
Arrays.fill(chars, 'f');
String s = new String(chars);
inputStream = new ByteArrayInputStream(s.getBytes());
}
@Test
void testSha256Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(SHA256, inputStream);
assertEquals("8098be8d5e22198e69316b1eef0b2863d75973b708f2ec4849132122b45fea71", generateChecksum);
assertEquals("b23c6c2faa06fe3f9a47b86914d08a77c9db9ce9b87fc8b6b2758db756159b2e", generateChecksum);
}
@Test
void testMd5Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(MD5, inputStream);
assertEquals("64c03752176a3a66e627db31ad4d3a9d", generateChecksum);
assertEquals("352b53342cc1ef2a21c480656ef6ffe6", generateChecksum);
}
@Test
void testSha384Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(SHA384, inputStream);
assertEquals("f7a88d89699d5fb9d4fb3b84b606bb1cac6641ed04e51f598fb7729e2142d5b0cec0c4019801d239749ff6ea0a872604", generateChecksum);
assertEquals("bd04277a8fc6ace52123d6d6214d2612ccdec802f361974198c44fd3df857b79f2343f010943340aeb7c51321a2d32e9", generateChecksum);
}
@Test
void testSha512Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(SHA512, inputStream);
assertEquals("5709738c21dfc8ea5b80a291a0e11c4c38bc8393d01b075e712877bf4b6329f822ab1a6d8eecd0d578e7ae9a51fff852f3c9849d23f5109a8068f92a858a88f0",
assertEquals("2efb6e7ef97eca294e8cc4bf731615622199fba59c2b5ed4f9a56c1f17be3522abac4f5d9fa0f95f39ff10a59c28597a263697b2d794e6686d260ffaa078da7e",
generateChecksum);
}
@Test
void testCrc32Checksum() {
String generateChecksum = ChecksumUtil.generateChecksum(CRC32, inputStream);
assertEquals("666124cf", generateChecksum);
assertEquals("4b8f2fda", generateChecksum);
}
@Test
void testMurmur3_32Checksum(){
assertDoesNotThrow(()->ChecksumUtil.generateChecksum(MURMUR3_32, inputStream));
void testMurmur3_32Checksum() {
assertEquals("3d45a1dd", ChecksumUtil.generateChecksum(MURMUR3_32, inputStream));
assertDoesNotThrow(() -> ChecksumUtil.generateChecksum(MURMUR3_32, inputStream));
}
@Test
void testMurmur3_128Checksum(){
assertDoesNotThrow(()->ChecksumUtil.generateChecksum(MURMUR3_128, inputStream));
void testMurmur3_128Checksum() {
assertEquals("9dbf0ffe5f6ecbb5edf207717e7870b7", ChecksumUtil.generateChecksum(MURMUR3_128, inputStream));
assertDoesNotThrow(() -> ChecksumUtil.generateChecksum(MURMUR3_128, inputStream));
}
}

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

@ -74,6 +74,7 @@ import org.thingsboard.server.dao.user.UserService;
import org.thingsboard.server.dao.widget.WidgetTypeService;
import org.thingsboard.server.dao.widget.WidgetsBundleService;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Comparator;
@ -269,7 +270,7 @@ public abstract class AbstractServiceTest {
firmware.setContentType("text/plain");
firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256);
firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a");
firmware.setData(BlobProxy.generateProxy(new byte[]{1}));
firmware.setData(new ByteArrayInputStream(new byte[]{1}));
firmware.setDataSize(1L);
return firmware;
}

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

@ -38,6 +38,7 @@ import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.dao.exception.DataValidationException;
import java.io.ByteArrayInputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
@ -111,7 +112,7 @@ public abstract class BaseDeviceProfileServiceTest extends AbstractServiceTest {
firmware.setContentType("text/plain");
firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256);
firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a");
firmware.setData(BlobProxy.generateProxy(new byte[]{1}));
firmware.setData(new ByteArrayInputStream(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

@ -42,6 +42,7 @@ import org.thingsboard.server.common.data.security.DeviceCredentialsType;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
import org.thingsboard.server.dao.exception.DataValidationException;
import java.io.ByteArrayInputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
@ -201,7 +202,7 @@ public abstract class BaseDeviceServiceTest extends AbstractServiceTest {
firmware.setContentType("text/plain");
firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256);
firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a");
firmware.setData(BlobProxy.generateProxy(new byte[]{1}));
firmware.setData(new ByteArrayInputStream(new byte[]{1}));
firmware.setDataSize(1L);
OtaPackage savedFirmware = otaPackageService.saveOtaPackage(firmware);
@ -236,7 +237,7 @@ public abstract class BaseDeviceServiceTest extends AbstractServiceTest {
firmware.setContentType("text/plain");
firmware.setChecksumAlgorithm(ChecksumAlgorithm.SHA256);
firmware.setChecksum("4bf5122f344554c53bde2ebb8cd2b7e3d1600ad631c385a5d7cce23c7785459a");
firmware.setData(BlobProxy.generateProxy(new byte[]{1}));
firmware.setData(new ByteArrayInputStream(new byte[]{1}));
firmware.setDataSize(1L);
OtaPackage savedFirmware = otaPackageService.saveOtaPackage(firmware);

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

@ -32,10 +32,7 @@ import org.thingsboard.server.dao.exception.DataValidationException;
import org.thingsboard.server.dao.ota.util.ChecksumUtil;
import javax.validation.ValidationException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.*;
import java.sql.Blob;
import java.util.ArrayList;
import java.util.Collections;
@ -54,7 +51,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 Blob DATA = BlobProxy.generateProxy(new byte[]{(int) DATA_SIZE});
private static final InputStream DATA = new ByteArrayInputStream(new byte[]{1});
private static final String URL = "http://firmware.test.org";
private final IdComparator<OtaPackageInfo> idComparator = new IdComparator<>();
@ -693,13 +690,6 @@ public abstract class BaseOtaPackageServiceTest extends AbstractServiceTest {
otaPackageService.saveOtaPackageInfo(firmwareInfo, true);
}
@Test
public void testGettingCorrectStreamWithOtaData() {
OtaPackage firmware = createFirmware(tenantId, "24687846");
InputStream stream = otaPackageService.getOtaDataStream(tenantId, firmware.getId());
assertEquals(firmware.getChecksum(), ChecksumUtil.generateChecksum(CHECKSUM_ALGORITHM, stream));
}
@Test
public void testGettingCorrectFileWithOtaData() {
OtaPackage firmware = createFirmware(tenantId, "24687846");

Write Preview
Loading…
Cancel
Save