diff --git a/.github/workflows/mega-linter.yml b/.github/workflows/mega-linter.yml new file mode 100644 index 00000000000..5d8ce213425 --- /dev/null +++ b/.github/workflows/mega-linter.yml @@ -0,0 +1,45 @@ +--- +name: MegaLinter + +on: + pull_request: + branches: [develop] + +env: + APPLY_FIXES: all + APPLY_FIXES_EVENT: pull_request + APPLY_FIXES_MODE: commit + PRINT_ALPACA: false + GITHUB_COMMENT_REPORTER: false + +concurrency: + group: ${{ github.ref }}-${{ github.workflow }} + cancel-in-progress: true + +jobs: + build: + name: MegaLinter + runs-on: ubuntu-20.04 + steps: + - name: Checkout Code + uses: actions/checkout@v2 + with: + token: ${{ secrets.PAT || secrets.GITHUB_TOKEN }} + fetch-depth: 0 + + - name: MegaLinter + id: ml + uses: megalinter/megalinter@v6 + env: + ENABLE: GROOVY + VALIDATE_ALL_CODEBASE: ${{ github.event_name == 'push' && github.ref == 'refs/heads/develop' }} + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + + - name: Archive artifacts + if: ${{ success() }} || ${{ failure() }} + uses: actions/upload-artifact@v2 + with: + name: MegaLinter reports + path: | + report + mega-linter.log diff --git a/CHANGELOG.md b/CHANGELOG.md index e3e8665c9a3..541ec06b072 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,38 @@ # Changelog -## v1.122.0 (07/07/2022) +## v1.123.0 (08/08/2022) + +### Features: +- [#4866](https://github.com/telstra/open-kilda/pull/4866) Added possibility to delete lagPort in case no lagPort on a switch (Issue: [#4729](https://github.com/telstra/open-kilda/issues/4729)) [**storm-topologies**] +- [#4877](https://github.com/telstra/open-kilda/pull/4877) Added query parameter to /api/v1/network/paths for limiting amount of computed paths +- [#4842](https://github.com/telstra/open-kilda/pull/4842) Add information about misconfigured rules into switch validation response (Issue: [#4834](https://github.com/telstra/open-kilda/issues/4834)) [**docs**] + +### Bug Fixes: +- [#4871](https://github.com/telstra/open-kilda/pull/4871) Fix swap-endpoint history recording - response timeout if flow update is disabled (Issue: [#4788](https://github.com/telstra/open-kilda/issues/4788)) +- [#4879](https://github.com/telstra/open-kilda/pull/4879) Added hex field, which should display correctly in Swagger UI (Issue: [#2120](https://github.com/telstra/open-kilda/issues/2120)) [**api**] +- [#4891](https://github.com/telstra/open-kilda/pull/4891) Added command context in ActionBolt + +### Improvements: +- [#4870](https://github.com/telstra/open-kilda/pull/4870) Reworked FL RuleManager response topic handling (Issue: [#4860](https://github.com/telstra/open-kilda/issues/4860)) [**floodlight**] +- [#4872](https://github.com/telstra/open-kilda/pull/4872) Stats topology cache notifies zookeeper about readiness only after initialization [**storm-topologies**] +- [#4875](https://github.com/telstra/open-kilda/pull/4875) Add static code analysis for Groovy +- [#4883](https://github.com/telstra/open-kilda/pull/4883) Added test relating to issue #4517 (Issues: [#4517](https://github.com/telstra/open-kilda/issues/4517) [#4517](https://github.com/telstra/open-kilda/issues/4517)) [**tests**] +- [#4887](https://github.com/telstra/open-kilda/pull/4887) Add test for limiting amount of computed paths #4877 (Issues: [#4877](https://github.com/telstra/open-kilda/issues/4877) [#4877](https://github.com/telstra/open-kilda/issues/4877)) [**tests**] +- [#4888](https://github.com/telstra/open-kilda/pull/4888) Test/deleting already deleted lag port [**tests**] +- [#4894](https://github.com/telstra/open-kilda/pull/4894) Set version of @types/geojson to "7946.0.8" [**gui**] +- [#4895](https://github.com/telstra/open-kilda/pull/4895) Add notification to templates +- [#4897](https://github.com/telstra/open-kilda/pull/4897) Remove commented test data (Issue: [#3896](https://github.com/telstra/open-kilda/issues/3896)) [**tests**] +- [#4898](https://github.com/telstra/open-kilda/pull/4898) Remove unnecessary MegaLinter options [**tests**] +- [#4862](https://github.com/telstra/open-kilda/pull/4862) Removed unused methods and constructors from FlowDto + +For the complete list of changes, check out [the commit log](https://github.com/telstra/open-kilda/compare/v1.122.0...v1.123.0). + +### Affected Components: +flow-hs, history, flow-monitor, fl, gui, swmanager, stats + +--- + +## v1.122.0 (21/07/2022) ### Features: - [#4848](https://github.com/telstra/open-kilda/pull/4848) Added vlan statistics field to FlowV2 CRUD api (Issue: [#4855](https://github.com/telstra/open-kilda/issues/4855)) diff --git a/README.md b/README.md index 44d5bcd20ab..1df38da3b38 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ --- KILDA CONTROLLER --- -[![Build Status](https://github.com/telstra/open-kilda/actions/workflows/unittest.yml/badge.svg)](https://github.com/telstra/open-kilda/actions/workflows/unittest.yml)[![Quality Gate](https://sonarcloud.io/api/project_badges/measure?project=org.openkilda%3Akilda-parent&metric=alert_status)](https://sonarcloud.io/dashboard?id=org.openkilda%3Akilda-parent) +[![Build Status](https://github.com/telstra/open-kilda/actions/workflows/unittest.yml/badge.svg)](https://github.com/telstra/open-kilda/actions/workflows/unittest.yml)[![Quality Gate](https://sonarcloud.io/api/project_badges/measure?project=org.openkilda%3Akilda-parent&metric=alert_status)](https://sonarcloud.io/dashboard?id=org.openkilda%3Akilda-parent)[![Mega-Linter](https://github.com/telstra/open-kilda/actions/workflows/MegaLinter/badge.svg?branch=main)](https://github.com/telstra/open-kilda/actions?query=workflow%3AMegaLinter+branch%3Amain) ## Introduction diff --git a/confd/templates/base-storm-topology/topology.properties.tmpl b/confd/templates/base-storm-topology/topology.properties.tmpl index 700cc71132b..0e5c781c693 100644 --- a/confd/templates/base-storm-topology/topology.properties.tmpl +++ b/confd/templates/base-storm-topology/topology.properties.tmpl @@ -113,6 +113,7 @@ local.execution.time = 3000 ####### # Path computation engine pce.network.strategy = {{ getv "/kilda_pce_network_strategy" }} +pce.max.path.count = {{ getv "/kilda_pce_max_path_count" }} ####### # cost strategy params # value added to path weight for each isl used in the same diversity group diff --git a/confd/templates/connecteddevices-topology/connecteddevices-topology.tmpl b/confd/templates/connecteddevices-topology/connecteddevices-topology.tmpl index 4a6d2511434..8112b7eb02d 100644 --- a/confd/templates/connecteddevices-topology/connecteddevices-topology.tmpl +++ b/confd/templates/connecteddevices-topology/connecteddevices-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_parallelism_level_new" }} diff --git a/confd/templates/floodlightrouter-topology/floodlightrouter-topology.tmpl b/confd/templates/floodlightrouter-topology/floodlightrouter-topology.tmpl index 6f33282f559..7ee7b267405 100644 --- a/confd/templates/floodlightrouter-topology/floodlightrouter-topology.tmpl +++ b/confd/templates/floodlightrouter-topology/floodlightrouter-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_parallelism_level_new" }} diff --git a/confd/templates/flowhs-topology/flowhs-topology.tmpl b/confd/templates/flowhs-topology/flowhs-topology.tmpl index 33c14bf76d7..6f736340af3 100644 --- a/confd/templates/flowhs-topology/flowhs-topology.tmpl +++ b/confd/templates/flowhs-topology/flowhs-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_flow_hs_parallelism" }} diff --git a/confd/templates/flowmonitoring-topology/flowmonitoring-topology.tmpl b/confd/templates/flowmonitoring-topology/flowmonitoring-topology.tmpl index 942960d901f..08f9d7131b2 100644 --- a/confd/templates/flowmonitoring-topology/flowmonitoring-topology.tmpl +++ b/confd/templates/flowmonitoring-topology/flowmonitoring-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_flow_monitoring_parallelism" }} diff --git a/confd/templates/history-topology/history-topology.tmpl b/confd/templates/history-topology/history-topology.tmpl index 34be5da355d..8838472e04e 100644 --- a/confd/templates/history-topology/history-topology.tmpl +++ b/confd/templates/history-topology/history-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_history_parallelism" }} diff --git a/confd/templates/isllatency-topology/isllatency-topology.tmpl b/confd/templates/isllatency-topology/isllatency-topology.tmpl index 4f406ed6c47..66bff75a922 100644 --- a/confd/templates/isllatency-topology/isllatency-topology.tmpl +++ b/confd/templates/isllatency-topology/isllatency-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_isl_latency_parallelism" }} diff --git a/confd/templates/makefile/makefile.tmpl b/confd/templates/makefile/makefile.tmpl index 465db57bca6..1446acc7c80 100644 --- a/confd/templates/makefile/makefile.tmpl +++ b/confd/templates/makefile/makefile.tmpl @@ -1,6 +1,9 @@ TASK := functionalTest STABLE_TAG := "stable" +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + build-base: {{if not (exists "/no_grpc_stub")}}build-grpc-stub {{end}}build-lock-keeper update-props docker/storm/lib {{if not (exists "/no_server42_server")}}build-server42dpdk{{end}} docker build -t kilda/base-ubuntu:latest docker/base/kilda-base-ubuntu/ docker build -t kilda/zookeeper:latest docker/zookeeper diff --git a/confd/templates/nbworker-topology/nbworker-topology.tmpl b/confd/templates/nbworker-topology/nbworker-topology.tmpl index 4016eb3f4c0..443015f6c6d 100644 --- a/confd/templates/nbworker-topology/nbworker-topology.tmpl +++ b/confd/templates/nbworker-topology/nbworker-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_parallelism_level_new" }} diff --git a/confd/templates/network-topology/network-topology.tmpl b/confd/templates/network-topology/network-topology.tmpl index c8e54231c79..ec4a2145620 100644 --- a/confd/templates/network-topology/network-topology.tmpl +++ b/confd/templates/network-topology/network-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: 2 diff --git a/confd/templates/opentsdb-topology/opentsdb-topology.tmpl b/confd/templates/opentsdb-topology/opentsdb-topology.tmpl index 1b649214772..314743241f1 100644 --- a/confd/templates/opentsdb-topology/opentsdb-topology.tmpl +++ b/confd/templates/opentsdb-topology/opentsdb-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_parallelism_level" }} diff --git a/confd/templates/ping-topology/ping-topology.tmpl b/confd/templates/ping-topology/ping-topology.tmpl index acc59e9ed14..9ebcfca770a 100644 --- a/confd/templates/ping-topology/ping-topology.tmpl +++ b/confd/templates/ping-topology/ping-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: 2 diff --git a/confd/templates/portstate-topology/portstate-topology.tmpl b/confd/templates/portstate-topology/portstate-topology.tmpl index 969c223f4c0..0035136808e 100644 --- a/confd/templates/portstate-topology/portstate-topology.tmpl +++ b/confd/templates/portstate-topology/portstate-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_parallelism_level" }} diff --git a/confd/templates/reroute-topology/reroute-topology.tmpl b/confd/templates/reroute-topology/reroute-topology.tmpl index aa2dc0ccca5..7c777a92929 100644 --- a/confd/templates/reroute-topology/reroute-topology.tmpl +++ b/confd/templates/reroute-topology/reroute-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_parallelism_level_new" }} diff --git a/confd/templates/server42/server42-control-topology.tmpl b/confd/templates/server42/server42-control-topology.tmpl index 7fe0783583a..86b7aa4b86c 100644 --- a/confd/templates/server42/server42-control-topology.tmpl +++ b/confd/templates/server42/server42-control-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_parallelism_level_new" }} diff --git a/confd/templates/stats-topology/stats-topology.tmpl b/confd/templates/stats-topology/stats-topology.tmpl index fb4466eb422..716f979474f 100644 --- a/confd/templates/stats-topology/stats-topology.tmpl +++ b/confd/templates/stats-topology/stats-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_stats_parallelism" }} diff --git a/confd/templates/swmanager-topology/swmanager-topology.tmpl b/confd/templates/swmanager-topology/swmanager-topology.tmpl index 1bf42f7a7c7..e6f853a3e55 100644 --- a/confd/templates/swmanager-topology/swmanager-topology.tmpl +++ b/confd/templates/swmanager-topology/swmanager-topology.tmpl @@ -1,3 +1,6 @@ +# Generated by confd. +# Do not change this file, all changes will be lost. Change corresponding template. + # topology configuration config: topology.parallelism: {{ getv "/kilda_storm_swmanager_parallelism" }} diff --git a/confd/vars/main.yaml b/confd/vars/main.yaml index 8af2f983405..a743f47151f 100644 --- a/confd/vars/main.yaml +++ b/confd/vars/main.yaml @@ -123,6 +123,7 @@ kilda_bfd_interval_ms: 350 kilda_bfd_multiplier: 3 kilda_pce_network_strategy: "SYMMETRIC_COST" +kilda_pce_max_path_count: 500 kilda_floodlight_alive_timeout: 10 kilda_floodlight_alive_interval: 2 diff --git a/docs/design/hub-and-spoke/switch-validate/switch-validate.md b/docs/design/hub-and-spoke/switch-validate/switch-validate.md index 8ff3dc2c5a2..6fda765ce60 100644 --- a/docs/design/hub-and-spoke/switch-validate/switch-validate.md +++ b/docs/design/hub-and-spoke/switch-validate/switch-validate.md @@ -5,9 +5,667 @@ ![Switch validation design](h&s-switch-validate.png "Switch validation design") ## FSM for validation switch rules and meters + Here is a FSM diagram that helps to understand main steps of validation switch rules and meters. ![Switch validation fsm](switch-validate-fsm.png "Switch validation fsm diagram") -### For more details about hub&spoke and look into examples please follow this [link](https://github.com/telstra/open-kilda/blob/develop/docs/design/hub-and-spoke/v7/README.md) +### For more details about hub&spoke and look into [examples](https://github.com/telstra/open-kilda/blob/develop/docs/design/hub-and-spoke/v7/README.md) + +[Issue #1551](https://github.com/telstra/open-kilda/issues/1551) + +## v1 API summary: +* [Validate rules installed on the switch](#validate-rules-installed-on-the-switch) + +`GET https://{host}/v1/switches/{switch_id}/rules/validate` + +* [Validate rules, meters, lags and groups installed on the switch v1](#validate-rules-meters-lags-and-groups-installed-on-the-switch-v1) + +`GET https://{host}/v1/switches/{switch_id}/validate` + +## v2 API summary: +* [Validate rules, meters, lags and groups installed on the switch v2](#validate-rules-meters-lags-and-groups-installed-on-the-switch-v2) + +`GET https://{host}/v2/switches/{switch_id}/validate` + +**Default: without any query params response contains v2 info about groups, lags, meters and rules.** + +* [Validate rules, meters, lags and groups installed on the switch with filters](#validate-rules-meters-lags-and-groups-installed-on-the-switch-with-filters) + +`GET https://{host}/v2/switches/{switch_id}/validate?include=groups|meters|logical_ports|rules&exclude=flow_info` + +## v1 API : GET + +### Validate rules installed on the switch + +`GET https://{host}/v1/switches/{switch_id}/rules/validate` + +Response payload + +```json +{ + "excess-rules-hex": [ + "8000000000000001" + ], + "excess_rules": [ + -9223372036854776000 + ], + "missing-rules-hex": [ + "800000000000001b" + ], + "missing_rules": [ + -9223372036854776000 + ], + "proper-rules-hex": [ + "8000000000000001" + ], + "proper_rules": [ + -9223372036854776000 + ] +} +``` + +### General info + +
v1GroupInfo + +```json +{ + "group_id": 10, + "excess_group_buckets": [ + { + "port": 2, + "vlan": 2, + "vni": 2 + } + ], + "group_buckets": [ + { + "port": 3, + "vlan": 3, + "vni": 3 + } + ], + "missing_group_buckets": [ + { + "port": 4, + "vlan": 4, + "vni": 4 + } + ] +} +``` + +
+ +
v1LogicalPortInfo + +```json +{ + "logical_port_number": 100, + "physical_ports": [ + 1 + ], + "type": "lag", + "actual": { // only for misconfigured section + "physical_ports": [ + 10, + 11 + ], + "type": "lag" + }, + "expected": { // only for misconfigured section + "physical_ports": [ + 10, + 11, + 13 + ], + "type": "bfd" + } +} +``` +
+ +
v1MeterInfo + +```json +{ + "rate": 300, + "burst_size": 200, + "cookie": 10, + "flow_id": "1231232", + "meter_id": 1, + "flags": [ + "STATS" + ], + "actual": { // only for misconfigured section + "flags": [ + "KBPS" + ], + "rate": 250, + "burst_size": 220 + }, + "expected": { // only for misconfigured section + "flags": [ + "STATS" + ], + "rate": 200, + "burst_size": 230 + } +} +``` +
+ +### Validate rules, meters, lags and groups installed on the switch v1 + +`GET https://{host}/v1/switches/{switch_id}/validate` + +Response payload + +```json +{ + "groups": { + "excess": [ + + ], + "misconfigured": [ + + ], + "missing": [ + + ], + "proper": [ + + ] + }, + "logical_ports": { + "error": "some_error", + "excess": [ + + ], + "misconfigured": [ + + ], + "missing": [ + + ], + "proper": [ + + ] + }, + "meters": { + "excess": [ + + ], + "misconfigured": [ + + ], + "missing": [ + + ], + "proper": [ + + ] + }, + "rules": { + "excess": [ + -9223372036854776000 + ], + "excess-hex": [ + "800000000000001b" + ], + "misconfigured": [ + -9223372036854776000 + ], + "misconfigured-hex": [ + "800000000000001b" + ], + "missing": [ + -9223372036854776000 + ], + "missing-hex": [ + "800000000000001b" + ], + "proper": [ + -9223372036854776000 + ], + "proper-hex": [ + "800000000000001b" + ] + } +} +``` + +## v2 API : GET + +### Groups, meters, lags and rules info include flow_id, flow_path and y_flow_id(for meters,groups and rules). + +
v2GroupInfo + +```json +{ + "flow_id": "10", + "flow_path": "e0b0716e-cc59-4ef6-8572-518bdcdc5c72", + "group_id": 10, + "buckets": [ + { + "port": 3, + "vlan": 3, + "vni": 3 + } + ] +} +``` + +
+ +
v2LogicalPortInfo + +```json +{ + "logical_port_number": 1001, + "type": "lag", + "physical_ports": [ + 1, + 2, + 3 + ] +} +``` +
+ +
v2MeterInfo + +```json +{ + "flow_id": "10", + "flow_path": "e0b0716e-cc59-4ef6-8572-518bdcdc5c72", + "y_flow_id": "111", + "meter_id": 1, + "flags": [ + "STATS" + ], + "rate": 300, + "burst_size": 200 +} +``` +
+ +
v2RuleInfo + +```json +{ + "cookie": -9223372036854776000, + "cookie_hex": "800000000000001b", + "cookie_kind": "LLDP_INPUT_PRE_DROP", // not implemented yet + "table_id": 10, + "priority": 5, + "flow_id": "9", + "flow_path": "10a1df3a-dc30-453e-8671-246f82ea0e77", + "y_flow_id": "111", + "flags": [ + "RESET_COUNTERS" + ], + "match": [ + { + "in_port": {"value": 1, "mask": 0, "is_masked": false} + } + ], + "instructions": { + "go_to_table": 30, + "go_to_meter": 40, + "write_metadata": { + "value": 200, + "mask": 200 + }, + "apply_actions": [ + + ], + "write_actions": [ + + ] + } +} +``` + +
+ +### Validate rules, meters, lags and groups installed on the switch v2 + +`GET https://{host}/v2/switches/{switch_id}/validate` + +**By Default:** + +* `flow info` is set of values: for groups - `flow_id, flow_path`, for meters and rules - `flow_id, flow_path, y_flow_id`. +* without any query params response contains v2 info about groups, lags, meters and rules (with flow_info). +* groups[excess], meters[excess] and rule[excess] do not contain flow info. +* for groups, lags, meters and rules flow info will be presented in misconfigured[expected] field. +* misconfigured[id] is + * meter_id for meters + * logical_port_id for logical_ports + * group_id for groups + * (priority + match).toString() for rules // may change during api implementation + +Response payload + +```json +{ + "as_expected": true, + "groups": { + "as_expected": true, + "excess": [ + // without flow_id and flow_path + ], + "missing": [ + + ], + "proper": [ + + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + + }, + "discrepancies": { + + } + } + ] + }, + "logical_ports": { + "as_expected": true, + "excess": [ + + ], + "missing": [ + + ], + "proper": [ + + ], + "misconfigured": [ + { + "id": "123123", + "expected": { + + }, + "discrepancies": { + + } + } + ] + }, + "meters": { + "as_expected": true, + "excess": [ + // without flow_id, y_flow_id and flow_path + ], + "missing": [ + + ], + "proper": [ + + ], + "misconfigured": [ + { + "id": "1232131", + "expected": { + + }, + "discrepancies": { + + } + } + ] + }, + "rules": { + "as_expected": true, + "excess": [ + // without flow_id, y_flow_id and flow_path + ], + "missing": [ + + ], + "proper": [ + + ], + "misconfigured": [ + { + "id": "123123123", + "expected": { + + }, + "discrepancies": { + + } + } + ] + } +} +``` + +### Validate rules, meters, lags and groups installed on the switch with filters + +### Requesting validation of specific switch objects/subjects. + +Request params + +To get only group info in response use query param `groups`, other fields will be omitted. + +`GET https://{host}/v2/switches/{switch_id}/validate?include=groups` + +Response payload + +```json +{ + "groups": { + "as_expected": true, + "excess": [ + // without flow_id and flow_path + ], + "missing": [ + + ], + "proper": [ + + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + + }, + "discrepancies": { + + } + } + ] + } +} +``` + +All other targets can be selected in same way i.e `include=meters` for meters, `include=rules` and +so on. + +You can combine several flags in one query e.g. to get full info only about groups and meters use query +params `groups` and `meters`. + +`GET https://{host}/v2/switches/{switch_id}/validate?include=groups|meters` + +### Excluding flow info from response + +Request params + +To exclude flow info from response use `flow_info`. + +`GET https://{host}/v2/switches/{switch_id}/validate?exclude=flow_info` + +
Response payload + +```json +{ + "as_expected": true, + "groups": { + "as_expected": true, + "excess": [ + // without flow_id and flow_path + ], + "missing": [ + // without flow_id and flow_path + ], + "proper": [ + // without flow_id and flow_path + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + // without flow_id and flow_path + }, + "discrepancies": { + // without flow_id and flow_path + } + } + ] + }, + "logical_ports": { + "as_expected": true, + "excess": [ + + ], + "missing": [ + + ], + "proper": [ + + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + + }, + "discrepancies": { + + } + } + ] + }, + "meters": { + "as_expected": true, + "excess": [ + // without flow_id, y_flow_id and flow_path + ], + "missing": [ + // without flow_id, y_flow_id and flow_path + ], + "proper": [ + // without flow_id, y_flow_id and flow_path + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + // without flow_id, y_flow_id and flow_path + }, + "discrepancies": { + // without flow_id, y_flow_id and flow_path + } + } + ] + }, + "rules": { + "as_expected": true, + "excess": [ + // without flow_id, y_flow_id and flow_path + ], + "missing": [ + // without flow_id, y_flow_id and flow_path + ], + "proper": [ + // without flow_id, y_flow_id and flow_path + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + // without flow_id, y_flow_id and flow_path + }, + "discrepancies": { + // without flow_id, y_flow_id and flow_path + } + } + ] + } +} +``` +
+ +### Including/excluding info in response + +To exclude flow info from response use `exclude=flow_info`. You can combine it with `include=groups|logical_ports|meters|rules` +in query e.g. + +Ex 1. `GET https://{host}/v2/switches/{switch_id}/validate?exclude=flow_info&include=groups|meters` + +In this case response will show you only group and meter info (without flow info) and omit other +validated fields. + +Response payload + +```json +{ + "as_expected": true, + "groups": { + "as_expected": true, + "excess": [ + // without flow_id and flow_path + ], + "missing": [ + // without flow_id and flow_path + ], + "proper": [ + // without flow_id and flow_path + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + // without flow_id and flow_path + }, + "discrepancies": { + // without flow_id and flow_path + } + } + ] + }, + "meters": { + "as_expected": true, + "excess": [ + // without flow_id, y_flow_id and flow_path + ], + "missing": [ + // without flow_id, y_flow_id and flow_path + ], + "proper": [ + // without flow_id, y_flow_id and flow_path + ], + "misconfigured": [ + { + "id": "12123213", + "expected": { + // without flow_id, y_flow_id and flow_path + }, + "discrepancies": { + // without flow_id, y_flow_id and flow_path + } + } + ] + } +} +``` + +Ex 2. `GET https://{host}/v2/switches/{switch_id}/validate?exclude=flow_info&include=rules` -[Issue #1551](https://github.com/telstra/open-kilda/issues/1551) \ No newline at end of file +In this case response will show you only rule info without flow info. diff --git a/src-gui/ui/package.json b/src-gui/ui/package.json index 159777539bf..d8fa796d391 100644 --- a/src-gui/ui/package.json +++ b/src-gui/ui/package.json @@ -23,6 +23,7 @@ "@ng-bootstrap/ng-bootstrap": "^3.2.0", "@ng-select/ng-select": "^2.8.0", "@types/d3": "^4.13.0", + "@types/geojson": "7946.0.8", "@types/jqueryui": "^1.12.4", "angular-datatables": "^6.0.0", "angular-font-awesome": "^3.1.2", diff --git a/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/model/FlowDto.java b/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/model/FlowDto.java index 4d9b6d84674..a500e153046 100644 --- a/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/model/FlowDto.java +++ b/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/model/FlowDto.java @@ -17,7 +17,6 @@ import org.openkilda.messaging.Utils; import org.openkilda.messaging.payload.flow.FlowEncapsulationType; -import org.openkilda.messaging.payload.flow.FlowPayload; import org.openkilda.messaging.payload.flow.FlowState; import org.openkilda.messaging.payload.flow.FlowStatusDetails; import org.openkilda.model.PathComputationStrategy; @@ -349,88 +348,6 @@ public FlowDto(@JsonProperty(Utils.FLOW_ID) final String flowId, this.vlanStatistics = vlanStatistics; } - /** - * Instance constructor. - * - * @param flowId flow id - * @param bandwidth bandwidth - * @param ignoreBandwidth ignore bandwidth flag - * @param description description - * @param sourceSwitch source switch - * @param sourcePort source port - * @param sourceVlan source vlan id - * @param destinationSwitch destination switch - * @param destinationPort destination port - * @param destinationVlan destination vlan id - * @param pinned pinned flag - * @param detectConnectedDevices detect connected devices flags - */ - public FlowDto(String flowId, - long bandwidth, - boolean ignoreBandwidth, - String description, - SwitchId sourceSwitch, int sourcePort, int sourceVlan, - SwitchId destinationSwitch, int destinationPort, int destinationVlan, boolean pinned, - DetectConnectedDevicesDto detectConnectedDevices) { - this(flowId, - bandwidth, - ignoreBandwidth, - false, - false, - false, - 0, - description, - null, null, - sourceSwitch, - destinationSwitch, - sourcePort, - destinationPort, - sourceVlan, - destinationVlan, 0, 0, - null, 0, null, null, null, null, null, null, pinned, null, detectConnectedDevices, null, null, null, - null, null, null, null, null, null, null, null, null); - } - - public FlowDto(FlowPayload input) { - this(input.getId(), - input.getMaximumBandwidth(), - input.isIgnoreBandwidth(), - false, - input.isPeriodicPings(), - input.isAllocateProtectedPath(), - 0, - input.getDescription(), - null, null, - input.getSource().getDatapath(), - input.getDestination().getDatapath(), - input.getSource().getPortNumber(), - input.getDestination().getPortNumber(), - input.getSource().getVlanId(), - input.getDestination().getVlanId(), - input.getSource().getInnerVlanId(), - input.getDestination().getInnerVlanId(), - null, 0, null, null, null, - input.getMaxLatency(), - null, - input.getPriority(), - input.isPinned(), - input.getEncapsulationType() != null ? FlowEncapsulationType.valueOf( - input.getEncapsulationType().toUpperCase()) : null, - new DetectConnectedDevicesDto( - input.getSource().getDetectConnectedDevices().isLldp(), - input.getSource().getDetectConnectedDevices().isArp(), - input.getDestination().getDetectConnectedDevices().isLldp(), - input.getDestination().getDetectConnectedDevices().isArp()), - input.getPathComputationStrategy() != null ? PathComputationStrategy.valueOf( - input.getPathComputationStrategy().toUpperCase()) : null, null, null, - null, null, null, null, null, null, null, null, null); - } - - @JsonIgnore - public long getFlagglessCookie() { - return cookie & MASK_COOKIE_FLAGS; - } - /** * Returns whether this represents a forward flow. * The result is based on the cookie value, diff --git a/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/payload/history/FlowDumpPayload.java b/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/payload/history/FlowDumpPayload.java index ffd7740d859..a6643c49bc6 100644 --- a/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/payload/history/FlowDumpPayload.java +++ b/src-java/base-topology/base-messaging/src/main/java/org/openkilda/messaging/payload/history/FlowDumpPayload.java @@ -42,8 +42,12 @@ public class FlowDumpPayload { private long forwardCookie; + private String forwardCookieHex; + private long reverseCookie; + private String reverseCookieHex; + private String sourceSwitch; private String destinationSwitch; diff --git a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java index d0a9005662e..3f4baa425f6 100644 --- a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java +++ b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/AbstractBolt.java @@ -182,9 +182,10 @@ protected final boolean shouldHandleLifeCycleEvent(Signal signal) { protected final void handleLifeCycleEvent(LifecycleEvent event) { if (Signal.START.equals(event.getSignal())) { - emit(ZkStreams.ZK.toString(), currentTuple, new Values(event, commandContext)); try { - activate(); + if (activateAndConfirm()) { + emit(ZkStreams.ZK.toString(), currentTuple, new Values(event, commandContext)); + } } finally { active = true; } @@ -206,6 +207,11 @@ protected void activate() { // no actions required } + protected boolean activateAndConfirm() { + activate(); + return true; + } + protected boolean deactivate(LifecycleEvent event) { return true; } @@ -268,7 +274,7 @@ protected CommandContext setupCommandContext() { context = new CommandContext().fork("trace-fail"); log.warn("The command context is missing in input tuple received by {} on stream {}:{}, execution context" - + " can't be traced. Create new command context for possible tracking of following" + + " can't be traced. Create new command context for possible tracking of following" + " processing [{}].", getClass().getName(), input.getSourceComponent(), input.getSourceStreamId(), formatTuplePayload(input), e); diff --git a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/share/mappers/HistoryMapper.java b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/share/mappers/HistoryMapper.java index 8c50cc94bd4..098d5b2feae 100644 --- a/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/share/mappers/HistoryMapper.java +++ b/src-java/base-topology/base-storm-topology/src/main/java/org/openkilda/wfm/share/mappers/HistoryMapper.java @@ -79,8 +79,13 @@ public abstract FlowHistoryEntry map( public FlowDumpPayload map(FlowEventDump dump) { FlowDumpPayload result = generatedMap(dump); - result.setForwardCookie(fallbackIfNull(mapCookie(dump.getForwardCookie()), 0L)); - result.setReverseCookie(fallbackIfNull(mapCookie(dump.getReverseCookie()), 0L)); + long forwardCookie = fallbackIfNull(mapCookie(dump.getForwardCookie()), 0L); + long reverseCookie = fallbackIfNull(mapCookie(dump.getReverseCookie()), 0L); + + result.setForwardCookie(forwardCookie); + result.setReverseCookie(reverseCookie); + result.setForwardCookieHex(Long.toHexString(forwardCookie)); + result.setReverseCookieHex(Long.toHexString(reverseCookie)); return result; } @@ -192,6 +197,8 @@ public String map(PathId pathId) { @Mapping(target = "forwardCookie", ignore = true) @Mapping(target = "reverseCookie", ignore = true) + @Mapping(target = "forwardCookieHex", ignore = true) + @Mapping(target = "reverseCookieHex", ignore = true) protected abstract FlowDumpPayload generatedMap(FlowEventDump dump); /** diff --git a/src-java/base-topology/base-storm-topology/src/test/java/org/openkilda/wfm/share/mappers/HistoryMapperTest.java b/src-java/base-topology/base-storm-topology/src/test/java/org/openkilda/wfm/share/mappers/HistoryMapperTest.java index fe04cad70e6..6f955d3d6c5 100644 --- a/src-java/base-topology/base-storm-topology/src/test/java/org/openkilda/wfm/share/mappers/HistoryMapperTest.java +++ b/src-java/base-topology/base-storm-topology/src/test/java/org/openkilda/wfm/share/mappers/HistoryMapperTest.java @@ -22,10 +22,14 @@ import org.openkilda.messaging.payload.history.FlowHistoryEntry; import org.openkilda.messaging.payload.history.FlowHistoryPayload; import org.openkilda.model.FlowEncapsulationType; +import org.openkilda.model.FlowPathStatus; +import org.openkilda.model.MeterId; import org.openkilda.model.PathComputationStrategy; import org.openkilda.model.SwitchId; +import org.openkilda.model.cookie.FlowSegmentCookie; import org.openkilda.model.history.FlowEvent; import org.openkilda.model.history.FlowEventAction; +import org.openkilda.model.history.FlowEventDump; import org.junit.BeforeClass; import org.junit.Test; @@ -46,34 +50,42 @@ public class HistoryMapperTest { public static int BANDWIDTH = 2; public static int DESTINATION_INNER_VLAN = 3; public static int DESTINATION_PORT = 4; - public static String DESTINATION_SWITCH = "5"; + public static String DESTINATION_SWITCH = "00:00:00:00:00:00:00:05"; public static int DESTINATION_VLAN = 6; public static String DIVERSE_GROUP_ID = "7"; public static FlowEncapsulationType FLOW_ENCAPSULATION_TYPE = FlowEncapsulationType.VXLAN; - public static int FORWARD_COOKIE = 8; + public static long FORWARD_COOKIE = 8L; public static Long FORWARD_METER_ID = 9L; public static String FORWARD_PATH = "10"; - public static String FORWARD_STATUS = "11"; + public static String FORWARD_STATUS = "ACTIVE"; public static boolean IGNORE_BANDWIDTH = false; public static SwitchId LOOP_SWITCH_ID = new SwitchId(12); - public static int MAX_LATENCY = 13; + public static long MAX_LATENCY = 13; public static PathComputationStrategy PATH_COMPUTATION_STRATEGY = PathComputationStrategy.COST; public static boolean PERIODIC_PINGS = false; public static boolean PINNED = false; - public static int REVERSE_COOKIE = 14; + public static long REVERSE_COOKIE = 14L; public static long REVERSE_METER_ID = 15; public static String REVERSE_PATH = "16"; - public static String REVERSE_STATUS = "17"; + public static String REVERSE_STATUS = "DEGRADED"; public static int SOURCE_INNER_VLAN = 18; public static int SOURCE_PORT = 19; - public static String SOURCE_SWITCH = "20"; + public static String SOURCE_SWITCH = "00:00:00:00:00:00:00:20"; public static int SOURCE_VLAN = 21; public static String TYPE = "22"; + public static long MAX_LATENCY_TIER_2 = 23; + + public static Integer PRIORITY = 24; + + public static boolean STRICT_BANDWIDTH = true; + public static FlowEventAction action; public static FlowEvent event; - public static FlowDumpPayload flowDumpPayload; + public static FlowDumpPayload expectedPayload; + + public static FlowEventDump flowEventDump; @BeforeClass public static void initializeData() { @@ -90,7 +102,8 @@ public static void initializeData() { .timestamp(TIMESTAMP) .actor(ACTOR) .build(); - flowDumpPayload = FlowDumpPayload.builder() + + expectedPayload = FlowDumpPayload.builder() .affinityGroupId(AFFINITY_GROUP_ID) .allocateProtectedPath(ALLOCATE_PROTECTED_PATH) .bandwidth(BANDWIDTH) @@ -101,6 +114,7 @@ public static void initializeData() { .diverseGroupId(DIVERSE_GROUP_ID) .encapsulationType(FLOW_ENCAPSULATION_TYPE) .forwardCookie(FORWARD_COOKIE) + .forwardCookieHex("8") .forwardMeterId(FORWARD_METER_ID) .forwardPath(FORWARD_PATH) .forwardStatus(FORWARD_STATUS) @@ -111,6 +125,7 @@ public static void initializeData() { .periodicPings(PERIODIC_PINGS) .pinned(PINNED) .reverseCookie(REVERSE_COOKIE) + .reverseCookieHex("e") .reverseMeterId(REVERSE_METER_ID) .reversePath(REVERSE_PATH) .reverseStatus(REVERSE_STATUS) @@ -119,7 +134,44 @@ public static void initializeData() { .sourceSwitch(SOURCE_SWITCH) .sourceVlan(SOURCE_VLAN) .type(TYPE) + .maxLatencyTier2(MAX_LATENCY_TIER_2) + .priority(PRIORITY) + .strictBandwidth(STRICT_BANDWIDTH) .build(); + + flowEventDump = new FlowEventDump(); + + flowEventDump.setAffinityGroupId(AFFINITY_GROUP_ID); + flowEventDump.setAllocateProtectedPath(ALLOCATE_PROTECTED_PATH); + flowEventDump.setBandwidth(BANDWIDTH); + flowEventDump.setDestinationInnerVlan(DESTINATION_INNER_VLAN); + flowEventDump.setDestinationPort(DESTINATION_PORT); + flowEventDump.setDestinationSwitch(new SwitchId(DESTINATION_SWITCH)); + flowEventDump.setDestinationVlan(DESTINATION_VLAN); + flowEventDump.setDiverseGroupId(DIVERSE_GROUP_ID); + flowEventDump.setEncapsulationType(FLOW_ENCAPSULATION_TYPE); + flowEventDump.setForwardCookie(new FlowSegmentCookie(FORWARD_COOKIE)); + flowEventDump.setForwardMeterId(new MeterId(FORWARD_METER_ID)); + flowEventDump.setForwardPath(FORWARD_PATH); + flowEventDump.setForwardStatus(FlowPathStatus.ACTIVE); + flowEventDump.setIgnoreBandwidth(IGNORE_BANDWIDTH); + flowEventDump.setLoopSwitchId(LOOP_SWITCH_ID); + flowEventDump.setMaxLatency(MAX_LATENCY); + flowEventDump.setPathComputationStrategy(PATH_COMPUTATION_STRATEGY); + flowEventDump.setPeriodicPings(PERIODIC_PINGS); + flowEventDump.setPinned(PINNED); + flowEventDump.setReverseCookie(new FlowSegmentCookie(REVERSE_COOKIE)); + flowEventDump.setReverseMeterId(new MeterId(REVERSE_METER_ID)); + flowEventDump.setReversePath(REVERSE_PATH); + flowEventDump.setReverseStatus(FlowPathStatus.DEGRADED); + flowEventDump.setSourceInnerVlan(SOURCE_INNER_VLAN); + flowEventDump.setSourcePort(SOURCE_PORT); + flowEventDump.setSourceSwitch(new SwitchId(SOURCE_SWITCH)); + flowEventDump.setSourceVlan(SOURCE_VLAN); + flowEventDump.setType(TYPE); + flowEventDump.setPriority(PRIORITY); + flowEventDump.setMaxLatencyTier2(MAX_LATENCY_TIER_2); + flowEventDump.setStrictBandwidth(STRICT_BANDWIDTH); } @Test @@ -140,7 +192,7 @@ public void toFlowHistoryEntryTest() { flowHistoryPayloads.add(flowHistoryPayload); ArrayList flowDumpPayloads = new ArrayList<>(); - flowDumpPayloads.add(flowDumpPayload); + flowDumpPayloads.add(expectedPayload); FlowHistoryEntry entry = INSTANCE.map(event, flowHistoryPayloads, flowDumpPayloads); @@ -155,4 +207,11 @@ public void toFlowHistoryEntryTest() { assertEquals(entry.getTimestampIso(), "2021-02-09T19:28:45Z"); assertEquals(entry.getTaskId(), event.getTaskId()); } + + @Test + public void toFlowDumpPayloadTest() { + FlowDumpPayload payload = INSTANCE.map(flowEventDump); + assertEquals(payload, expectedPayload); + + } } diff --git a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/BatchCommandProcessor.java b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/BatchCommandProcessor.java index 156b9e3af6b..3e97717eea6 100644 --- a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/BatchCommandProcessor.java +++ b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/BatchCommandProcessor.java @@ -18,7 +18,6 @@ import org.openkilda.floodlight.api.request.rulemanager.DeleteSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.ModifySpeakerCommandsRequest; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse; public interface BatchCommandProcessor { @@ -29,5 +28,5 @@ public interface BatchCommandProcessor { void processBatchDelete(DeleteSpeakerCommandsRequest request, String key); - void processResponse(SpeakerCommandResponse response, String key, Origin origin); + void processResponse(SpeakerCommandResponse response, String key, String sourceTopic); } diff --git a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/BaseSpeakerCommandsRequest.java b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/BaseSpeakerCommandsRequest.java index 07c08b5ff02..0949252af3b 100644 --- a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/BaseSpeakerCommandsRequest.java +++ b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/BaseSpeakerCommandsRequest.java @@ -42,6 +42,7 @@ @Type(value = DeleteSpeakerCommandsRequest.class, name = "org.openkilda.floodlight.api.request.rulemanager.DeleteSpeakerCommandsRequest") }) + @Getter @ToString(callSuper = true) public abstract class BaseSpeakerCommandsRequest extends SpeakerRequest { @@ -49,25 +50,20 @@ public abstract class BaseSpeakerCommandsRequest extends SpeakerRequest { @JsonProperty("command_data") protected Collection commands; - @JsonProperty("origin") - protected Origin origin; + @JsonProperty("source_topic") + protected String sourceTopic; public BaseSpeakerCommandsRequest(MessageContext messageContext, @NonNull SwitchId switchId, @NonNull UUID commandId, - Collection commands, - Origin origin) { + Collection commands) { + super(messageContext, switchId, commandId); this.commands = commands; - this.origin = origin; - } - - public Collection getCommands() { - return commands; } - public Origin getOrigin() { - return origin; + public void setSourceTopic(String sourceTopic) { + this.sourceTopic = sourceTopic; } public abstract void process(BatchCommandProcessor processor, String key); diff --git a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/DeleteSpeakerCommandsRequest.java b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/DeleteSpeakerCommandsRequest.java index feeef43094f..ce9125408e1 100644 --- a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/DeleteSpeakerCommandsRequest.java +++ b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/DeleteSpeakerCommandsRequest.java @@ -34,9 +34,8 @@ public class DeleteSpeakerCommandsRequest extends BaseSpeakerCommandsRequest { public DeleteSpeakerCommandsRequest(@JsonProperty("message_context") MessageContext messageContext, @JsonProperty("switch_id") @NonNull SwitchId switchId, @JsonProperty("command_id") @NonNull UUID commandId, - @JsonProperty("command_data") Collection commands, - @JsonProperty("origin") Origin origin) { - super(messageContext, switchId, commandId, commands, origin); + @JsonProperty("command_data") Collection commands) { + super(messageContext, switchId, commandId, commands); } public void process(BatchCommandProcessor processor, String key) { diff --git a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/InstallSpeakerCommandsRequest.java b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/InstallSpeakerCommandsRequest.java index 0337c5e6a20..b4cfdb0d6d5 100644 --- a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/InstallSpeakerCommandsRequest.java +++ b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/InstallSpeakerCommandsRequest.java @@ -40,9 +40,8 @@ public InstallSpeakerCommandsRequest(@JsonProperty("message_context") MessageCon @JsonProperty("switch_id") @NonNull SwitchId switchId, @JsonProperty("command_id") @NonNull UUID commandId, @JsonProperty("command_data") Collection commands, - @JsonProperty("origin") Origin origin, @JsonProperty("fail_if_exists") Boolean failIfExists) { - super(messageContext, switchId, commandId, commands, origin); + super(messageContext, switchId, commandId, commands); this.failIfExists = failIfExists == null || failIfExists; } diff --git a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/ModifySpeakerCommandsRequest.java b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/ModifySpeakerCommandsRequest.java index 524e1dbfead..618e3611eb5 100644 --- a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/ModifySpeakerCommandsRequest.java +++ b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/ModifySpeakerCommandsRequest.java @@ -34,9 +34,8 @@ public class ModifySpeakerCommandsRequest extends BaseSpeakerCommandsRequest { public ModifySpeakerCommandsRequest(@JsonProperty("message_context") MessageContext messageContext, @JsonProperty("switch_id") @NonNull SwitchId switchId, @JsonProperty("command_id") @NonNull UUID commandId, - @JsonProperty("command_data") Collection commands, - @JsonProperty("origin") Origin origin) { - super(messageContext, switchId, commandId, commands, origin); + @JsonProperty("command_data") Collection commands) { + super(messageContext, switchId, commandId, commands); } public void process(BatchCommandProcessor processor, String key) { diff --git a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/Origin.java b/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/Origin.java deleted file mode 100644 index f55312668d8..00000000000 --- a/src-java/floodlight-service/floodlight-api/src/main/java/org/openkilda/floodlight/api/request/rulemanager/Origin.java +++ /dev/null @@ -1,23 +0,0 @@ -/* Copyright 2021 Telstra Open Source - * - * 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. - */ - -package org.openkilda.floodlight.api.request.rulemanager; - -public enum Origin { - - FLOW_HS, - SW_MANAGER, - NETWORK -} diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/KafkaChannel.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/KafkaChannel.java index e19646d4a51..af3b3985aa1 100644 --- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/KafkaChannel.java +++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/KafkaChannel.java @@ -80,7 +80,7 @@ public String getSpeakerTopic() { return formatTopicWithRegion(topics.getSpeakerRegionTopic()); } - public String getSpeakerFlowTopic() { + public String getSpeakerFlowHsTopic() { return formatTopicWithRegion(topics.getSpeakerFlowRegionTopic()); } @@ -128,7 +128,7 @@ public String getTopoSwitchManagerTopic() { return formatTopicWithRegion(topics.getTopoSwitchManagerRegionTopic()); } - public String getSpeakerFlowHsTopic() { + public String getSpeakerFlowHsResponseTopic() { return formatTopicWithRegion(topics.getFlowHsSpeakerRegionTopic()); } diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/flow/FlowSegmentReport.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/flow/FlowSegmentReport.java index 127c9c6324e..72fa4c8c6df 100644 --- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/flow/FlowSegmentReport.java +++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/flow/FlowSegmentReport.java @@ -47,7 +47,8 @@ protected FlowSegmentReport(@NonNull FlowSegmentCommand command, Exception error } public void reply(KafkaChannel kafkaChannel, IKafkaProducerService kafkaProducerService, String requestKey) { - kafkaProducerService.sendMessageAndTrack(kafkaChannel.getSpeakerFlowHsTopic(), requestKey, assembleResponse()); + kafkaProducerService.sendMessageAndTrack(kafkaChannel.getSpeakerFlowHsResponseTopic(), + requestKey, assembleResponse()); } private SpeakerResponse assembleResponse() { diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java index 4fbdbdb8340..196be852895 100644 --- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java +++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutor.java @@ -22,7 +22,6 @@ import static java.util.stream.Collectors.groupingBy; import org.openkilda.floodlight.api.BatchCommandProcessor; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.floodlight.service.session.Session; import org.openkilda.floodlight.service.session.SessionService; import org.openkilda.messaging.MessageContext; @@ -62,8 +61,8 @@ public class OfBatchExecutor { private final OfBatchHolder holder; private final Set switchFeatures; private final String kafkaKey; - private final Origin origin; private final boolean failIfExists; + private final String sourceTopic; private boolean hasMeters; private boolean hasGroups; @@ -76,7 +75,8 @@ public class OfBatchExecutor { @Builder public OfBatchExecutor(IOFSwitch iofSwitch, BatchCommandProcessor commandProcessor, SessionService sessionService, MessageContext messageContext, OfBatchHolder holder, - Set switchFeatures, String kafkaKey, Origin origin, Boolean failIfExists) { + Set switchFeatures, String kafkaKey, String sourceTopic, + Boolean failIfExists) { this.iofSwitch = iofSwitch; this.commandProcessor = commandProcessor; this.sessionService = sessionService; @@ -90,7 +90,7 @@ public OfBatchExecutor(IOFSwitch iofSwitch, BatchCommandProcessor commandProcess this.holder = holder; this.switchFeatures = switchFeatures; this.kafkaKey = kafkaKey; - this.origin = origin; + this.sourceTopic = sourceTopic; this.failIfExists = failIfExists == null || failIfExists; } @@ -343,6 +343,6 @@ private void verifyGroups() { } private void sendResponse() { - commandProcessor.processResponse(holder.getResult(), kafkaKey, origin); + commandProcessor.processResponse(holder.getResult(), kafkaKey, sourceTopic); } } diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfSpeakerService.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfSpeakerService.java index 1cd1e7636b6..83156802bf8 100644 --- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfSpeakerService.java +++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/command/rulemanager/OfSpeakerService.java @@ -24,7 +24,6 @@ import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.ModifySpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse; import org.openkilda.floodlight.service.FeatureDetectorService; import org.openkilda.floodlight.service.kafka.IKafkaProducerService; @@ -32,6 +31,7 @@ import org.openkilda.floodlight.service.session.SessionService; import org.openkilda.model.SwitchId; +import com.google.common.collect.ImmutableMap; import edu.umd.cs.findbugs.annotations.NonNull; import lombok.extern.slf4j.Slf4j; import net.floodlightcontroller.core.IOFSwitch; @@ -39,20 +39,31 @@ import net.floodlightcontroller.core.module.FloodlightModuleContext; import org.projectfloodlight.openflow.types.DatapathId; +import java.util.Map; + @Slf4j public class OfSpeakerService implements BatchCommandProcessor { private final IOFSwitchService iofSwitchService; private final SessionService sessionService; - private final KafkaUtilityService kafkaUtilityService; private final IKafkaProducerService kafkaProducerService; private final FeatureDetectorService featureDetectorService; + private final Map responseTopics; + public OfSpeakerService(@NonNull FloodlightModuleContext moduleContext) { this.iofSwitchService = moduleContext.getServiceImpl(IOFSwitchService.class); this.sessionService = moduleContext.getServiceImpl(SessionService.class); - this.kafkaUtilityService = moduleContext.getServiceImpl(KafkaUtilityService.class); this.kafkaProducerService = moduleContext.getServiceImpl(IKafkaProducerService.class); this.featureDetectorService = moduleContext.getServiceImpl(FeatureDetectorService.class); + + KafkaUtilityService kafkaUtilityService = moduleContext.getServiceImpl(KafkaUtilityService.class); + KafkaChannel kafkaChannel = kafkaUtilityService.getKafkaChannel(); + + responseTopics = ImmutableMap.of( + kafkaChannel.getSpeakerFlowHsTopic(), kafkaChannel.getSpeakerFlowHsResponseTopic(), + kafkaChannel.getNetworkControlTopic(), kafkaChannel.getNetworkControlResponseTopic(), + kafkaChannel.getSpeakerSwitchManagerTopic(), kafkaChannel.getSpeakerSwitchManagerResponseTopic() + ); } @Override @@ -82,7 +93,7 @@ private void processBatchRequest(BaseSpeakerCommandsRequest request, String key, } if (sw == null) { log.warn("Switch {} not found. Can't process request {}.", switchId, request); - processResponse(holder.getResult(), key, request.getOrigin()); + processResponse(holder.getResult(), key, request.getSourceTopic()); return; } OfBatchExecutor executor = OfBatchExecutor.builder() @@ -93,8 +104,8 @@ private void processBatchRequest(BaseSpeakerCommandsRequest request, String key, .holder(holder) .switchFeatures(featureDetectorService.detectSwitch(sw)) .kafkaKey(key) - .origin(request.getOrigin()) .failIfExists(failIfExists) + .sourceTopic(request.getSourceTopic()) .build(); executor.executeBatch(); } @@ -112,25 +123,13 @@ private void buildDeleteOfCommand(OfCommand ofCommand, OfBatchHolder holder, Swi } @Override - public void processResponse(SpeakerCommandResponse response, String kafkaKey, Origin origin) { - KafkaChannel kafkaChannel = kafkaUtilityService.getKafkaChannel(); - String topic = getTopic(kafkaChannel, origin); - log.debug("Send response to {} (key={})", topic, kafkaKey); - kafkaProducerService.sendMessageAndTrack(topic, kafkaKey, response); - } - - private String getTopic(KafkaChannel kafkaChannel, Origin origin) { - //TODO: remove origin and detect response topic by income topic - switch (origin) { - case FLOW_HS: - return kafkaChannel.getSpeakerFlowHsTopic(); - case SW_MANAGER: - return kafkaChannel.getSpeakerSwitchManagerResponseTopic(); - case NETWORK: - return kafkaChannel.getNetworkControlResponseTopic(); - default: - throw new IllegalStateException(format("Unknown message origin %s", origin)); + public void processResponse(SpeakerCommandResponse response, String kafkaKey, String sourceTopic) { + String responseTopic = responseTopics.get(sourceTopic); + if (responseTopic == null) { + throw new IllegalStateException(format("Unknown message sourceTopic %s", sourceTopic)); } + log.debug("Send response to {} (key={})", responseTopic, kafkaKey); + kafkaProducerService.sendMessageAndTrack(responseTopic, kafkaKey, response); } @FunctionalInterface diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/ConsumerContext.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/ConsumerContext.java index e587e92b7c3..880c959d4ab 100644 --- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/ConsumerContext.java +++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/ConsumerContext.java @@ -89,6 +89,6 @@ public String getKafkaSwitchManagerTopic() { } public String getKafkaSpeakerFlowHsTopic() { - return kafkaChannel.getSpeakerFlowHsTopic(); + return kafkaChannel.getSpeakerFlowHsResponseTopic(); } } diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/KafkaMessageCollector.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/KafkaMessageCollector.java index 4352b6f6f8a..f4d692a9f26 100644 --- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/KafkaMessageCollector.java +++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/KafkaMessageCollector.java @@ -100,7 +100,7 @@ protected void launchTopics(KafkaMessageCollectorConfig consumerConfig, ExecutorService generalExecutor = buildExecutorWithNoQueue(consumerConfig.getGeneralExecutorCount()); logger.info("Kafka Consumer: general executor threads = {}", consumerConfig.getGeneralExecutorCount()); launcher.launch(generalExecutor, new KafkaConsumerSetup(kafkaChannel.getSpeakerTopic())); - launcher.launch(generalExecutor, new KafkaConsumerSetup(kafkaChannel.getSpeakerFlowTopic())); + launcher.launch(generalExecutor, new KafkaConsumerSetup(kafkaChannel.getSpeakerFlowHsTopic())); launcher.launch(generalExecutor, new KafkaConsumerSetup(kafkaChannel.getSpeakerSwitchManagerTopic(), kafkaChannel.getNetworkControlTopic())); launcher.launch(generalExecutor, new KafkaConsumerSetup(kafkaChannel.getSpeakerFlowPingTopic())); diff --git a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java index 94d1ce23827..4100d2f1173 100644 --- a/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java +++ b/src-java/floodlight-service/floodlight-modules/src/main/java/org/openkilda/floodlight/kafka/RecordHandler.java @@ -770,7 +770,7 @@ private void parseRecord(ConsumerRecord record) { if (handleSpeakerCommand()) { return; } - if (handleRuleManagerCommand()) { + if (handleRuleManagerCommand(record.topic())) { return; } @@ -829,7 +829,7 @@ private void handleSpeakerCommand(SpeakerCommand } } - private boolean handleRuleManagerCommand() { + private boolean handleRuleManagerCommand(String sourceTopic) { BaseSpeakerCommandsRequest request; try { request = MAPPER.readValue(record.value(), BaseSpeakerCommandsRequest.class); @@ -841,6 +841,7 @@ private boolean handleRuleManagerCommand() { return false; } + request.setSourceTopic(sourceTopic); try { handleRuleManagerCommand(request); } catch (Exception e) { diff --git a/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutorTest.java b/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutorTest.java index 06a09dbe087..cff145b87b5 100644 --- a/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutorTest.java +++ b/src-java/floodlight-service/floodlight-modules/src/test/java/org/openkilda/floodlight/command/rulemanager/OfBatchExecutorTest.java @@ -24,7 +24,6 @@ import static org.mockito.Mockito.when; import org.openkilda.floodlight.api.BatchCommandProcessor; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse; import org.openkilda.floodlight.service.session.Session; import org.openkilda.floodlight.service.session.SessionService; @@ -75,7 +74,7 @@ public class OfBatchExecutorTest { .holder(holder) .switchFeatures(Collections.emptySet()) .kafkaKey("kafka-key") - .origin(Origin.SW_MANAGER) + .sourceTopic("flowhs-topic") .build(); @Test @@ -103,7 +102,7 @@ public void shouldSendSuccessResponse() { executor.executeBatch(); ArgumentCaptor captor = ArgumentCaptor.forClass(SpeakerCommandResponse.class); - verify(batchCommandProcessor).processResponse(captor.capture(), any(String.class), any(Origin.class)); + verify(batchCommandProcessor).processResponse(captor.capture(), any(String.class), any(String.class)); assertTrue(captor.getValue().isSuccess()); verifyNoMoreInteractions(batchCommandProcessor); @@ -133,7 +132,7 @@ public void shouldSendFailedResponse() { executor.executeBatch(); ArgumentCaptor captor = ArgumentCaptor.forClass(SpeakerCommandResponse.class); - verify(batchCommandProcessor).processResponse(captor.capture(), any(String.class), any(Origin.class)); + verify(batchCommandProcessor).processResponse(captor.capture(), any(String.class), any(String.class)); assertFalse(captor.getValue().isSuccess()); verifyNoMoreInteractions(batchCommandProcessor); diff --git a/src-java/floodlightrouter-topology/floodlightrouter-storm-topology/src/main/java/org/openkilda/wfm/topology/floodlightrouter/FloodlightRouterTopology.java b/src-java/floodlightrouter-topology/floodlightrouter-storm-topology/src/main/java/org/openkilda/wfm/topology/floodlightrouter/FloodlightRouterTopology.java index d3475cddec6..9aec333ec20 100644 --- a/src-java/floodlightrouter-topology/floodlightrouter-storm-topology/src/main/java/org/openkilda/wfm/topology/floodlightrouter/FloodlightRouterTopology.java +++ b/src-java/floodlightrouter-topology/floodlightrouter-storm-topology/src/main/java/org/openkilda/wfm/topology/floodlightrouter/FloodlightRouterTopology.java @@ -166,7 +166,7 @@ private void networkToSpeaker(TopologyBuilder topology, TopologyOutput output) { makeRegionTopics(kafkaTopics.getNetworkControlResponseRegionTopic()), ComponentType.KILDA_NETWORK_REQUEST_KAFKA_SPOUT); declareSpeakerToControllerProxy( - topology, kafkaTopics.getNetworkControlReponseTopic(), + topology, kafkaTopics.getNetworkControlResponseTopic(), ComponentType.KILDA_NETWORK_REQUEST_KAFKA_SPOUT, ComponentType.KILDA_NETWORK_RESPONSE_REPLY_BOLT, output.getKafkaHsOutput()); diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java index b475fd72439..28dcf27c923 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/FlowProcessingWithHistorySupportFsm.java @@ -51,18 +51,26 @@ protected FlowProcessingWithHistorySupportFsm(@NonNull E nextEvent, @NonNull E e super(nextEvent, errorEvent, commandContext, carrier, eventListeners); } + protected String[] getFlowIdsForHistory() { + return new String[]{getFlowId()}; + } + /** * Add a history record on the action. */ public void saveActionToHistory(String action) { - log.debug("Flow {} action - {}", getFlowId(), action); - sendHistoryData(action, null); + for (String flowId : getFlowIdsForHistory()) { + log.debug("Flow {} action - {}", flowId, action); + sendHistoryData(flowId, action, null); + } } @Override public void saveActionToHistory(String action, String description) { - log.debug("Flow {} action - {} : {}", getFlowId(), action, description); - sendHistoryData(action, description); + for (String flowId : getFlowIdsForHistory()) { + log.debug("Flow {} action - {} : {}", flowId, action, description); + sendHistoryData(flowId, action, description); + } } /** @@ -85,26 +93,32 @@ public void saveFlowActionToHistory(String flowId, String action, String descrip @Override public void saveErrorToHistory(String action, String errorMessage) { - log.error("Flow {} error - {} : {}", getFlowId(), action, errorMessage); - sendHistoryData(action, errorMessage); + for (String flowId : getFlowIdsForHistory()) { + log.error("Flow {} error - {} : {}", flowId, action, errorMessage); + sendHistoryData(flowId, action, errorMessage); + } } @Override public void saveErrorToHistory(String errorMessage) { - log.error("Flow {} error - {}", getFlowId(), errorMessage); - sendHistoryData(errorMessage, null); + for (String flowId : getFlowIdsForHistory()) { + log.error("Flow {} error - {}", flowId, errorMessage); + sendHistoryData(flowId, errorMessage, null); + } } /** * Add a history record on the error. */ public void saveErrorToHistory(String errorMessage, Exception ex) { - log.error("Flow {} error - {}", getFlowId(), errorMessage, ex); - sendHistoryData(errorMessage, null); + for (String flowId : getFlowIdsForHistory()) { + log.error("Flow {} error - {}", flowId, errorMessage, ex); + sendHistoryData(flowId, errorMessage, null); + } } - protected void sendHistoryData(String action, String description) { - sendHistoryData(getFlowId(), action, description, getCommandContext().getCorrelationId()); + protected void sendHistoryData(String flowId, String action, String description) { + sendHistoryData(flowId, action, description, getCommandContext().getCorrelationId()); } protected void sendHistoryData(String flowId, String action, String description, String taskId) { @@ -141,7 +155,9 @@ public void saveNewEventToHistory(String flowId, String action, FlowEventData.Ev public void saveNewEventToHistory(String action, FlowEventData.Event event, FlowEventData.Initiator initiator, String details) { - saveNewEventToHistory(getFlowId(), action, event, initiator, details, getCommandContext().getCorrelationId()); + for (String flowId : getFlowIdsForHistory()) { + saveNewEventToHistory(flowId, action, event, initiator, details, getCommandContext().getCorrelationId()); + } } /** @@ -176,19 +192,21 @@ public void saveNewEventToHistory(String flowId, String action, FlowEventData.Ev */ public void saveActionWithDumpToHistory(String action, String description, FlowDumpData flowDumpData) { - log.debug("Flow {} action - {} : {}", getFlowId(), action, description); - - FlowHistoryHolder historyHolder = FlowHistoryHolder.builder() - .taskId(getCommandContext().getCorrelationId()) - .flowDumpData(flowDumpData) - .flowHistoryData(FlowHistoryData.builder() - .action(action) - .time(getNextHistoryEntryTime()) - .description(description) - .flowId(getFlowId()) - .build()) - .build(); - getCarrier().sendHistoryUpdate(historyHolder); + for (String flowId : getFlowIdsForHistory()) { + log.debug("Flow {} action - {} : {}", flowId, action, description); + + FlowHistoryHolder historyHolder = FlowHistoryHolder.builder() + .taskId(getCommandContext().getCorrelationId()) + .flowDumpData(flowDumpData) + .flowHistoryData(FlowHistoryData.builder() + .action(action) + .time(getNextHistoryEntryTime()) + .description(description) + .flowId(flowId) + .build()) + .build(); + getCarrier().sendHistoryUpdate(historyHolder); + } } public final Instant getNextHistoryEntryTime() { @@ -205,8 +223,10 @@ public final Instant getNextHistoryEntryTime() { } public void saveGlobalTimeoutToHistory() { - saveErrorToHistory(String.format( - "Global timeout reached for %s operation on flow \"%s\"", getCrudActionName(), getFlowId())); + for (String flowId : getFlowIdsForHistory()) { + saveErrorToHistory(String.format( + "Global timeout reached for %s operation on flow \"%s\"", getCrudActionName(), flowId)); + } } protected abstract String getCrudActionName(); diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/YFlowRuleManagerProcessingAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/YFlowRuleManagerProcessingAction.java index 68df213afab..3a588b07682 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/YFlowRuleManagerProcessingAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/actions/YFlowRuleManagerProcessingAction.java @@ -22,7 +22,6 @@ import org.openkilda.floodlight.api.request.rulemanager.FlowCommand; import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.model.FlowPath; import org.openkilda.model.PathId; import org.openkilda.model.SwitchId; @@ -64,12 +63,12 @@ protected YFlowRuleManagerProcessingAction(PersistenceManager persistenceManager protected Collection buildYFlowInstallRequests(YFlow yFlow, CommandContext context) { Map> speakerData = buildYFlowSpeakerData(yFlow); - return FlowRulesConverter.INSTANCE.buildFlowInstallCommands(speakerData, context, Origin.FLOW_HS); + return FlowRulesConverter.INSTANCE.buildFlowInstallCommands(speakerData, context); } protected Collection buildYFlowDeleteRequests(YFlow yFlow, CommandContext context) { Map> speakerData = buildYFlowSpeakerData(yFlow); - return FlowRulesConverter.INSTANCE.buildFlowDeleteCommands(speakerData, context, Origin.FLOW_HS); + return FlowRulesConverter.INSTANCE.buildFlowDeleteCommands(speakerData, context); } private Map> buildYFlowSpeakerData(YFlow yFlow) { diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/converters/FlowRulesConverter.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/converters/FlowRulesConverter.java index 626ea543405..ddf87d9a3af 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/converters/FlowRulesConverter.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/common/converters/FlowRulesConverter.java @@ -20,7 +20,6 @@ import org.openkilda.floodlight.api.request.rulemanager.DeleteSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.messaging.MessageContext; import org.openkilda.model.SwitchId; import org.openkilda.rulemanager.SpeakerData; @@ -46,11 +45,11 @@ public final class FlowRulesConverter { * Build a list of InstallSpeakerCommandsRequest from the provided speakerData. */ public Collection buildFlowInstallCommands( - Map> speakerData, CommandContext context, Origin origin) { + Map> speakerData, CommandContext context) { return speakerData.entrySet().stream() .map(entry -> { List ofCommands = OfCommandConverter.INSTANCE.toOfCommands(entry.getValue()); - return buildFlowInstallCommand(entry.getKey(), ofCommands, context, origin); + return buildFlowInstallCommand(entry.getKey(), ofCommands, context); }) .collect(Collectors.toList()); } @@ -59,7 +58,7 @@ public Collection buildFlowInstallCommands( * Build a InstallSpeakerCommandsRequest from the provided OF commands. */ public InstallSpeakerCommandsRequest buildFlowInstallCommand(SwitchId switchId, List ofCommands, - CommandContext context, Origin origin) { + CommandContext context) { UUID commandId = commandIdGenerator.generate(); MessageContext messageContext = new MessageContext(commandId.toString(), context.getCorrelationId()); return InstallSpeakerCommandsRequest.builder() @@ -67,7 +66,6 @@ public InstallSpeakerCommandsRequest buildFlowInstallCommand(SwitchId switchId, .switchId(switchId) .commandId(commandId) .commands(ofCommands) - .origin(origin) .build(); } @@ -76,12 +74,12 @@ public InstallSpeakerCommandsRequest buildFlowInstallCommand(SwitchId switchId, * NOTICE: the given dependencies are reversed as required for deletion. */ public Collection buildFlowDeleteCommands( - Map> speakerData, CommandContext context, Origin origin) { + Map> speakerData, CommandContext context) { return speakerData.entrySet().stream() .map(entry -> { List ofCommands = OfCommandConverter.INSTANCE.toOfCommands(entry.getValue()); ofCommands = OfCommandConverter.INSTANCE.reverseDependenciesForDeletion(ofCommands); - return buildFlowDeleteCommand(entry.getKey(), ofCommands, context, origin); + return buildFlowDeleteCommand(entry.getKey(), ofCommands, context); }) .collect(toList()); } @@ -90,9 +88,9 @@ public Collection buildFlowDeleteCommands( * Build a DeleteSpeakerCommandsRequest from the provided OF commands. */ public DeleteSpeakerCommandsRequest buildFlowDeleteCommand(SwitchId switchId, List ofCommands, - CommandContext context, Origin origin) { + CommandContext context) { UUID commandId = commandIdGenerator.generate(); MessageContext messageContext = new MessageContext(commandId.toString(), context.getCorrelationId()); - return new DeleteSpeakerCommandsRequest(messageContext, switchId, commandId, ofCommands, origin); + return new DeleteSpeakerCommandsRequest(messageContext, switchId, commandId, ofCommands); } } diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/FlowSwapEndpointsFsm.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/FlowSwapEndpointsFsm.java index 85b49d9c7ad..80460c47ea2 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/FlowSwapEndpointsFsm.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/FlowSwapEndpointsFsm.java @@ -84,6 +84,11 @@ public String getFlowId() { throw new UnsupportedOperationException("Not implemented for swap flow endpoints operation. Skipping"); } + @Override + protected String[] getFlowIdsForHistory() { + return new String[]{firstFlowId, secondFlowId}; + } + @Override protected String getCrudActionName() { return "swap-endpoints"; diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/actions/ValidateFlowsAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/actions/ValidateFlowsAction.java index ab3211b195b..7fe710184ed 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/actions/ValidateFlowsAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/swapendpoints/actions/ValidateFlowsAction.java @@ -58,7 +58,9 @@ protected void perform(State from, State to, Event event, FlowSwapEndpointsConte RequestedFlow secondTargetFlow = stateMachine.getSecondTargetFlow(); if (!featureTogglesRepository.getOrDefault().getUpdateFlowEnabled()) { - throw new FlowProcessingException(ErrorType.NOT_PERMITTED, "Flow update feature is disabled"); + stateMachine.fireValidationError( + new ErrorData(ErrorType.NOT_PERMITTED, FlowSwapEndpointsFsm.GENERIC_ERROR_MESSAGE, + "Flow update feature is disabled")); } try { diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/RevertSharedEndpointRulesAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/RevertSharedEndpointRulesAction.java index 3c21aed9b26..bba28bc2cd1 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/RevertSharedEndpointRulesAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/RevertSharedEndpointRulesAction.java @@ -20,7 +20,6 @@ import org.openkilda.floodlight.api.request.rulemanager.DeleteSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.messaging.error.ErrorType; import org.openkilda.model.SwitchId; import org.openkilda.model.YFlow; @@ -61,7 +60,7 @@ protected void perform(State from, State to, Event event, YFlowPathSwapContext c stateMachine.getOldPrimaryPaths()); InstallSpeakerCommandsRequest installRequest = FlowRulesConverter.INSTANCE.buildFlowInstallCommand(sharedEndpoint, installOfCommands, - stateMachine.getCommandContext(), Origin.FLOW_HS); + stateMachine.getCommandContext()); stateMachine.addInstallSpeakerCommand(installRequest.getCommandId(), installRequest); List deleteOfCommands = stateMachine.getInstallNewYFlowOfCommands(); @@ -69,7 +68,7 @@ protected void perform(State from, State to, Event event, YFlowPathSwapContext c if (deleteOfCommands != null) { deleteOfCommands = OfCommandConverter.INSTANCE.reverseDependenciesForDeletion(deleteOfCommands); deleteRequest = FlowRulesConverter.INSTANCE.buildFlowDeleteCommand(sharedEndpoint, deleteOfCommands, - stateMachine.getCommandContext(), Origin.FLOW_HS); + stateMachine.getCommandContext()); stateMachine.addDeleteSpeakerCommand(deleteRequest.getCommandId(), deleteRequest); } diff --git a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/UpdateSharedEndpointRulesAction.java b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/UpdateSharedEndpointRulesAction.java index 33961ea1535..eefb61a368b 100644 --- a/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/UpdateSharedEndpointRulesAction.java +++ b/src-java/flowhs-topology/flowhs-storm-topology/src/main/java/org/openkilda/wfm/topology/flowhs/fsm/yflow/pathswap/actions/UpdateSharedEndpointRulesAction.java @@ -20,7 +20,6 @@ import org.openkilda.floodlight.api.request.rulemanager.DeleteSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.messaging.error.ErrorType; import org.openkilda.model.SwitchId; import org.openkilda.model.YFlow; @@ -63,14 +62,14 @@ protected void perform(State from, State to, Event event, YFlowPathSwapContext c stateMachine.setInstallNewYFlowOfCommands(installOfCommands); InstallSpeakerCommandsRequest installRequest = FlowRulesConverter.INSTANCE.buildFlowInstallCommand(sharedEndpoint, installOfCommands, - stateMachine.getCommandContext(), Origin.FLOW_HS); + stateMachine.getCommandContext()); stateMachine.addInstallSpeakerCommand(installRequest.getCommandId(), installRequest); List deleteOfCommands = stateMachine.getDeleteOldYFlowOfCommands(); deleteOfCommands = OfCommandConverter.INSTANCE.reverseDependenciesForDeletion(deleteOfCommands); DeleteSpeakerCommandsRequest deleteRequest = FlowRulesConverter.INSTANCE.buildFlowDeleteCommand(sharedEndpoint, deleteOfCommands, - stateMachine.getCommandContext(), Origin.FLOW_HS); + stateMachine.getCommandContext()); stateMachine.addDeleteSpeakerCommand(deleteRequest.getCommandId(), deleteRequest); if (stateMachine.getInstallSpeakerRequests().isEmpty() && stateMachine.getDeleteSpeakerRequests().isEmpty()) { diff --git a/src-java/flowmonitoring-topology/flowmonitoring-storm-topology/src/main/java/org/openkilda/wfm/topology/flowmonitoring/bolt/ActionBolt.java b/src-java/flowmonitoring-topology/flowmonitoring-storm-topology/src/main/java/org/openkilda/wfm/topology/flowmonitoring/bolt/ActionBolt.java index 2cacbd35227..b86b72dd9c6 100644 --- a/src-java/flowmonitoring-topology/flowmonitoring-storm-topology/src/main/java/org/openkilda/wfm/topology/flowmonitoring/bolt/ActionBolt.java +++ b/src-java/flowmonitoring-topology/flowmonitoring-storm-topology/src/main/java/org/openkilda/wfm/topology/flowmonitoring/bolt/ActionBolt.java @@ -120,7 +120,7 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields(FIELD_ID_PAYLOAD, FIELD_ID_CONTEXT)); declarer.declareStream(FLOW_HS_STREAM_ID.name(), new Fields(FIELD_ID_PAYLOAD, FIELD_ID_CONTEXT)); declarer.declareStream(FLOW_STATS_STREAM_ID.name(), - new Fields(FLOW_ID_FIELD, FLOW_DIRECTION_FIELD, LATENCY_FIELD)); + new Fields(FLOW_ID_FIELD, FLOW_DIRECTION_FIELD, LATENCY_FIELD, FIELD_ID_CONTEXT)); declarer.declareStream(ZkStreams.ZK.toString(), new Fields(ZooKeeperBolt.FIELD_ID_STATE, ZooKeeperBolt.FIELD_ID_CONTEXT)); } @@ -140,6 +140,7 @@ public void sendFlowRerouteRequest(String flowId) { @Override public void persistFlowStats(String flowId, String direction, long latency) { - emit(FLOW_STATS_STREAM_ID.name(), getCurrentTuple(), new Values(flowId, direction, latency)); + emit(FLOW_STATS_STREAM_ID.name(), getCurrentTuple(), new Values(flowId, direction, latency, + getCommandContext())); } } diff --git a/src-java/kilda-configuration/src/main/java/org/openkilda/config/KafkaTopicsConfig.java b/src-java/kilda-configuration/src/main/java/org/openkilda/config/KafkaTopicsConfig.java index 0f3cf9e2f57..0a64d20b952 100644 --- a/src-java/kilda-configuration/src/main/java/org/openkilda/config/KafkaTopicsConfig.java +++ b/src-java/kilda-configuration/src/main/java/org/openkilda/config/KafkaTopicsConfig.java @@ -121,7 +121,7 @@ public interface KafkaTopicsConfig { @Key("speaker.network.control.response.priv") @Default("kilda.speaker.network.control.response.priv") - String getNetworkControlReponseTopic(); + String getNetworkControlResponseTopic(); @Key("speaker.network.priv.region") @Default("kilda.network.control.response.priv.region") diff --git a/src-java/kilda-pce/src/main/java/org/openkilda/pce/PathComputerConfig.java b/src-java/kilda-pce/src/main/java/org/openkilda/pce/PathComputerConfig.java index 8b4f4eb4391..487b0c30b1c 100644 --- a/src-java/kilda-pce/src/main/java/org/openkilda/pce/PathComputerConfig.java +++ b/src-java/kilda-pce/src/main/java/org/openkilda/pce/PathComputerConfig.java @@ -24,6 +24,10 @@ @Configuration @Key("pce") public interface PathComputerConfig extends Serializable { + @Key("max.path.count") + @Default("500") + int getMaxPathCount(); + @Key("max.allowed.depth") @Default("35") int getMaxAllowedDepth(); diff --git a/src-java/nbworker-topology/nbworker-messaging/src/main/java/org/openkilda/messaging/nbtopology/request/GetPathsRequest.java b/src-java/nbworker-topology/nbworker-messaging/src/main/java/org/openkilda/messaging/nbtopology/request/GetPathsRequest.java index 0a70512e274..7be53db87d4 100644 --- a/src-java/nbworker-topology/nbworker-messaging/src/main/java/org/openkilda/messaging/nbtopology/request/GetPathsRequest.java +++ b/src-java/nbworker-topology/nbworker-messaging/src/main/java/org/openkilda/messaging/nbtopology/request/GetPathsRequest.java @@ -49,17 +49,22 @@ public class GetPathsRequest extends BaseRequest { @JsonProperty("max_latency_tier2") Duration maxLatencyTier2; + @JsonProperty("max_path_count") + Integer maxPathCount; + public GetPathsRequest(@JsonProperty("src_switch_id") SwitchId srcSwitchId, @JsonProperty("dst_switch_id") SwitchId dstSwitchId, @JsonProperty("encapsulation_type") FlowEncapsulationType encapsulationType, @JsonProperty("path_computation_strategy") PathComputationStrategy pathComputationStrategy, @JsonProperty("max_latency") Duration maxLatency, - @JsonProperty("max_latency_tier2") Duration maxLatencyTier2) { + @JsonProperty("max_latency_tier2") Duration maxLatencyTier2, + @JsonProperty("max_path_count") Integer maxPathCount) { this.srcSwitchId = srcSwitchId; this.dstSwitchId = dstSwitchId; this.encapsulationType = encapsulationType; this.pathComputationStrategy = pathComputationStrategy; this.maxLatency = maxLatency; this.maxLatencyTier2 = maxLatencyTier2; + this.maxPathCount = maxPathCount; } } diff --git a/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/bolts/PathsBolt.java b/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/bolts/PathsBolt.java index 135c7e20667..f87cfa2f05b 100644 --- a/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/bolts/PathsBolt.java +++ b/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/bolts/PathsBolt.java @@ -66,7 +66,7 @@ private List getPaths(GetPathsRequest request) { try { return pathService.getPaths(request.getSrcSwitchId(), request.getDstSwitchId(), request.getEncapsulationType(), request.getPathComputationStrategy(), request.getMaxLatency(), - request.getMaxLatencyTier2()); + request.getMaxLatencyTier2(), request.getMaxPathCount()); } catch (IllegalArgumentException e) { throw new MessageException(ErrorType.DATA_INVALID, e.getMessage(), "Bad request."); } catch (RecoverableException e) { diff --git a/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/PathsService.java b/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/PathsService.java index 3a3c2d0ab27..2c122bb1f10 100644 --- a/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/PathsService.java +++ b/src-java/nbworker-topology/nbworker-storm-topology/src/main/java/org/openkilda/wfm/topology/nbworker/services/PathsService.java @@ -46,7 +46,7 @@ @Slf4j public class PathsService { - public static final int MAX_PATH_COUNT = 500; + private final int defaultMaxPathCount; private PathComputer pathComputer; private SwitchRepository switchRepository; private SwitchPropertiesRepository switchPropertiesRepository; @@ -59,6 +59,7 @@ public PathsService(RepositoryFactory repositoryFactory, PathComputerConfig path PathComputerFactory pathComputerFactory = new PathComputerFactory( pathComputerConfig, new AvailableNetworkFactory(pathComputerConfig, repositoryFactory)); pathComputer = pathComputerFactory.getPathComputer(); + defaultMaxPathCount = pathComputerConfig.getMaxPathCount(); } /** @@ -66,8 +67,8 @@ public PathsService(RepositoryFactory repositoryFactory, PathComputerConfig path */ public List getPaths( SwitchId srcSwitchId, SwitchId dstSwitchId, FlowEncapsulationType requestEncapsulationType, - PathComputationStrategy requestPathComputationStrategy, Duration maxLatency, Duration maxLatencyTier2) - throws RecoverableException, SwitchNotFoundException, UnroutableFlowException { + PathComputationStrategy requestPathComputationStrategy, Duration maxLatency, Duration maxLatencyTier2, + Integer maxPathCount) throws RecoverableException, SwitchNotFoundException, UnroutableFlowException { if (Objects.equals(srcSwitchId, dstSwitchId)) { throw new IllegalArgumentException( String.format("Source and destination switch IDs are equal: '%s'", srcSwitchId)); @@ -103,7 +104,16 @@ public List getPaths( PathComputationStrategy pathComputationStrategy = Optional.ofNullable(requestPathComputationStrategy) .orElse(kildaConfiguration.getPathComputationStrategy()); - List flowPaths = pathComputer.getNPaths(srcSwitchId, dstSwitchId, MAX_PATH_COUNT, flowEncapsulationType, + + if (maxPathCount == null) { + maxPathCount = defaultMaxPathCount; + } + + if (maxPathCount <= 0) { + throw new IllegalArgumentException(String.format("Incorrect maxPathCount: %s", maxPathCount)); + } + + List flowPaths = pathComputer.getNPaths(srcSwitchId, dstSwitchId, maxPathCount, flowEncapsulationType, pathComputationStrategy, maxLatency, maxLatencyTier2); return flowPaths.stream().map(PathMapper.INSTANCE::map) diff --git a/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/PathsServiceTest.java b/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/PathsServiceTest.java index 45c92dcd33f..3f2fe5de6a0 100644 --- a/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/PathsServiceTest.java +++ b/src-java/nbworker-topology/nbworker-storm-topology/src/test/java/org/openkilda/wfm/topology/nbworker/services/PathsServiceTest.java @@ -17,12 +17,13 @@ import static junit.framework.TestCase.assertEquals; import static junit.framework.TestCase.assertTrue; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; import static org.openkilda.model.FlowEncapsulationType.TRANSIT_VLAN; import static org.openkilda.model.FlowEncapsulationType.VXLAN; import static org.openkilda.model.PathComputationStrategy.COST; import static org.openkilda.model.PathComputationStrategy.LATENCY; import static org.openkilda.model.PathComputationStrategy.MAX_LATENCY; -import static org.openkilda.wfm.topology.nbworker.services.PathsService.MAX_PATH_COUNT; import org.openkilda.config.provider.PropertiesBasedConfigurationProvider; import org.openkilda.messaging.info.network.PathsInfoData; @@ -56,6 +57,7 @@ import java.util.Optional; public class PathsServiceTest extends InMemoryGraphBasedTest { + private static final int MAX_PATH_COUNT = 500; private static final int SWITCH_COUNT = MAX_PATH_COUNT + 50; private static final int VXLAN_SWITCH_COUNT = MAX_PATH_COUNT / 2; @@ -134,7 +136,8 @@ public void init() { @Test public void findNPathsByTransitVlanAndCost() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, COST, null, null); + List paths = pathsService.getPaths( + SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, COST, null, null, MAX_PATH_COUNT); assertEquals(MAX_PATH_COUNT, paths.size()); assertPathLength(paths); @@ -153,8 +156,8 @@ public void findNPathsByTransitVlanAndCost() @Test public void findNPathsByTransitVlanIgnoreMaxLatency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - List paths = pathsService.getPaths( - SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, LATENCY, Duration.ofNanos(1L), Duration.ofNanos(2L)); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, LATENCY, + Duration.ofNanos(1L), Duration.ofNanos(2L), MAX_PATH_COUNT); assertTransitVlanAndLatencyPaths(paths); } @@ -163,23 +166,23 @@ public void findNPathsByVxlanAndEnoughMaxLatencyZeroTier2Latency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { Duration maxLatency = Duration.ofNanos(MIN_LATENCY * 2 + 1); List paths = pathsService.getPaths( - SWITCH_ID_1, SWITCH_ID_2, VXLAN, MAX_LATENCY, maxLatency, Duration.ZERO); + SWITCH_ID_1, SWITCH_ID_2, VXLAN, MAX_LATENCY, maxLatency, Duration.ZERO, MAX_PATH_COUNT); assertMaxLatencyPaths(paths, maxLatency, 1, VXLAN); } @Test public void findNPathsByTransitVlanAndTooSmallMaxLatency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - List paths = pathsService.getPaths( - SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, MAX_LATENCY, Duration.ofNanos(1L), Duration.ZERO); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, + MAX_LATENCY, Duration.ofNanos(1L), Duration.ZERO, MAX_PATH_COUNT); assertTrue(paths.isEmpty()); } @Test public void findNPathsByTransitVlanAndTooSmallMaxLatencyAndTier2Latency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - List paths = pathsService.getPaths( - SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, MAX_LATENCY, Duration.ofNanos(1L), Duration.ofNanos(2L)); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, + MAX_LATENCY, Duration.ofNanos(1L), Duration.ofNanos(2L), MAX_PATH_COUNT); assertTrue(paths.isEmpty()); } @@ -187,8 +190,8 @@ public void findNPathsByTransitVlanAndTooSmallMaxLatencyAndTier2Latency() public void findNPathsByTransitVlanAndTooSmallMaxLatencyEnoughTier2Latency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { Duration maxLatencyTier2 = Duration.ofNanos(MIN_LATENCY * 2 + 1); - List paths = pathsService.getPaths( - SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, MAX_LATENCY, Duration.ofNanos(1L), maxLatencyTier2); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, MAX_LATENCY, + Duration.ofNanos(1L), maxLatencyTier2, MAX_PATH_COUNT); assertMaxLatencyPaths(paths, maxLatencyTier2, 1, TRANSIT_VLAN); } @@ -197,7 +200,7 @@ public void findNPathsByTransitVlanAndZeroMaxLatency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { // as max latency param is 0 LATENCY starategy will be used instead. It means all 500 paths will be returned List paths = pathsService.getPaths( - SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, MAX_LATENCY, Duration.ZERO, null); + SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, MAX_LATENCY, Duration.ZERO, null, MAX_PATH_COUNT); assertTransitVlanAndLatencyPaths(paths); } @@ -206,34 +209,59 @@ public void findNPathsByTransitVlanAndNullMaxLatency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { // as max latency param is null LATENCY starategy will be used instead. It means all 500 paths will be returned List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, MAX_LATENCY, null, - null); + null, MAX_PATH_COUNT); assertTransitVlanAndLatencyPaths(paths); } @Test public void findNPathsByTransitVlanAndLatency() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, LATENCY, null, null); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, LATENCY, + null, null, MAX_PATH_COUNT); assertTransitVlanAndLatencyPaths(paths); } @Test public void findNPathsByVxlanAndCost() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, VXLAN, COST, null, null); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, VXLAN, COST, null, + null, MAX_PATH_COUNT); assertVxlanAndCostPathes(paths); } + @Test + public void findNPathsByMaxPathCount() + throws UnroutableFlowException, SwitchNotFoundException, RecoverableException { + int maxPathCount = 3; + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, VXLAN, COST, null, + null, maxPathCount); + assertThat(paths.size(), equalTo(maxPathCount)); + } + + @Test(expected = IllegalArgumentException.class) + public void checkMaxPathCountWhenFindNPaths() + throws UnroutableFlowException, SwitchNotFoundException, RecoverableException { + int maxPathCount = -1; + pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, VXLAN, COST, null, null, maxPathCount); + } + + @Test(expected = IllegalArgumentException.class) + public void checkForZeroMaxPathCountWhenFindNPaths() + throws UnroutableFlowException, SwitchNotFoundException, RecoverableException { + int maxPathCount = 0; + pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, VXLAN, COST, null, null, maxPathCount); + } + @Test(expected = IllegalArgumentException.class) public void findNPathsByVxlanSrcWithoutVxlanSupport() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - pathsService.getPaths(SWITCH_ID_3, SWITCH_ID_2, VXLAN, COST, null, null); + pathsService.getPaths(SWITCH_ID_3, SWITCH_ID_2, VXLAN, COST, null, null, MAX_PATH_COUNT); } @Test(expected = IllegalArgumentException.class) public void findNPathsByVxlanDstWithoutVxlanSupport() throws SwitchNotFoundException, RecoverableException, UnroutableFlowException { - pathsService.getPaths(SWITCH_ID_2, SWITCH_ID_3, VXLAN, COST, null, null); + pathsService.getPaths(SWITCH_ID_2, SWITCH_ID_3, VXLAN, COST, null, null, MAX_PATH_COUNT); } @Test @@ -244,7 +272,8 @@ public void findNPathsByTransitVlanAndDefaultStrategy() config.setPathComputationStrategy(LATENCY); }); // find N paths without strategy - List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, null, null, null); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, TRANSIT_VLAN, + null, null, null, MAX_PATH_COUNT); assertTransitVlanAndLatencyPaths(paths); } @@ -256,7 +285,8 @@ public void findNPathsByDefaultEncapsulationAndCost() config.setFlowEncapsulationType(VXLAN); }); // find N paths without encapsulation type - List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, null, COST, null, null); + List paths = pathsService.getPaths(SWITCH_ID_1, SWITCH_ID_2, null, COST, + null, null, MAX_PATH_COUNT); assertVxlanAndCostPathes(paths); } diff --git a/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/NetworkTopology.java b/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/NetworkTopology.java index 430a86c08ca..74aeb067941 100644 --- a/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/NetworkTopology.java +++ b/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/NetworkTopology.java @@ -156,7 +156,7 @@ private void inputSwitchManager(TopologyBuilder topology) { private void inputSpeakerControl(TopologyBuilder topology) { declareKafkaSpoutForAbstractMessage(topology, - kafkaTopics.getNetworkControlReponseTopic(), ComponentId.INPUT_SPEAKER_RULES.toString()); + kafkaTopics.getNetworkControlResponseTopic(), ComponentId.INPUT_SPEAKER_RULES.toString()); } private void workerSpeakerRules(TopologyBuilder topology) { diff --git a/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/storm/bolt/isl/IslHandler.java b/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/storm/bolt/isl/IslHandler.java index 90b456beade..9e4694eb8de 100644 --- a/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/storm/bolt/isl/IslHandler.java +++ b/src-java/network-topology/network-storm-topology/src/main/java/org/openkilda/wfm/topology/network/storm/bolt/isl/IslHandler.java @@ -19,7 +19,6 @@ import org.openkilda.floodlight.api.request.rulemanager.DeleteSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse; import org.openkilda.messaging.MessageContext; import org.openkilda.messaging.command.CommandData; @@ -211,7 +210,6 @@ public void sendIslRulesInstallCommand(SwitchId switchId, UUID commandId, List getPaths( @ApiParam(value = "Second tier for flow path latency in milliseconds. If there is no path with required " + "max_latency, max_latency_tier2 with be used instead. Used only with MAX_LATENCY strategy. " + "Other strategies will ignore this parameter.") - @RequestParam(value = "max_latency_tier2", required = false) - Long maxLatencyTier2Ms) { + @RequestParam(value = "max_latency_tier2", required = false) Long maxLatencyTier2Ms, + @ApiParam(value = "Maximum count of paths which will be calculated. " + + "If maximum path count is not specified, default value from Kilda Configuration will be used") + @RequestParam(value = "max_path_count", required = false) Integer maxPathCount) { Duration maxLatency = maxLatencyMs != null ? Duration.ofMillis(maxLatencyMs) : null; Duration maxLatencyTier2 = maxLatencyTier2Ms != null ? Duration.ofMillis(maxLatencyTier2Ms) : null; return networkService.getPaths(srcSwitchId, dstSwitchId, encapsulationType, pathComputationStrategy, maxLatency, - maxLatencyTier2); + maxLatencyTier2, maxPathCount); } /** diff --git a/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/NetworkService.java b/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/NetworkService.java index c4f0ec3aec4..ccda5b49503 100644 --- a/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/NetworkService.java +++ b/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/NetworkService.java @@ -33,5 +33,6 @@ public interface NetworkService { */ CompletableFuture getPaths( SwitchId srcSwitch, SwitchId dstSwitch, FlowEncapsulationType encapsulationType, - PathComputationStrategy pathComputationStrategy, Duration maxLatencyMs, Duration maxLatencyTier2); + PathComputationStrategy pathComputationStrategy, Duration maxLatencyMs, Duration maxLatencyTier2, + Integer maxPathCount); } diff --git a/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/impl/NetworkServiceImpl.java b/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/impl/NetworkServiceImpl.java index 7b912a8e8e8..af440555717 100644 --- a/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/impl/NetworkServiceImpl.java +++ b/src-java/northbound-service/northbound/src/main/java/org/openkilda/northbound/service/impl/NetworkServiceImpl.java @@ -57,9 +57,15 @@ public class NetworkServiceImpl implements NetworkService { @Override public CompletableFuture getPaths( SwitchId srcSwitch, SwitchId dstSwitch, FlowEncapsulationType encapsulationType, - PathComputationStrategy pathComputationStrategy, Duration maxLatency, Duration maxLatencyTier2) { + PathComputationStrategy pathComputationStrategy, Duration maxLatency, Duration maxLatencyTier2, + Integer maxPathCount) { String correlationId = RequestCorrelationId.getId(); + if (maxPathCount != null && maxPathCount <= 0) { + throw new MessageException(correlationId, System.currentTimeMillis(), ErrorType.PARAMETERS_INVALID, + "Bad max_path_count parameter", "The number MAX_PATH_COUNT should be positive"); + } + if (PathComputationStrategy.MAX_LATENCY.equals(pathComputationStrategy) && maxLatency == null) { throw new MessageException(correlationId, System.currentTimeMillis(), ErrorType.PARAMETERS_INVALID, "Missed max_latency parameter.", "MAX_LATENCY path computation strategy requires non null " @@ -68,7 +74,7 @@ public CompletableFuture getPaths( } GetPathsRequest request = new GetPathsRequest(srcSwitch, dstSwitch, encapsulationType, pathComputationStrategy, - maxLatency, maxLatencyTier2); + maxLatency, maxLatencyTier2, maxPathCount); CommandMessage message = new CommandMessage(request, System.currentTimeMillis(), correlationId); return messagingChannel.sendAndGetChunked(nbworkerTopic, message) diff --git a/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/bolts/CacheBolt.java b/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/bolts/CacheBolt.java index 967b9ac2db7..a10e3d5b68d 100644 --- a/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/bolts/CacheBolt.java +++ b/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/bolts/CacheBolt.java @@ -18,6 +18,7 @@ import static org.openkilda.wfm.topology.stats.StatsTopology.STATS_FIELD; import static org.openkilda.wfm.topology.utils.KafkaRecordTranslator.FIELD_ID_PAYLOAD; +import org.openkilda.bluegreen.LifecycleEvent; import org.openkilda.messaging.info.InfoData; import org.openkilda.messaging.info.InfoMessage; import org.openkilda.messaging.info.stats.FlowStatsData; @@ -59,11 +60,22 @@ public CacheBolt(PersistenceManager persistenceManager, String lifeCycleEventSou @PersistenceContextRequired(requiresNew = true) protected void init() { cacheService = new KildaEntryCacheService(persistenceManager, this); + } + + @Override + protected boolean activateAndConfirm() { try { - cacheService.refreshCache(); + cacheService.activate(); } catch (Exception ex) { - log.error("Error on сache initialization", ex); + log.error(String.format("Error on cache initialization: %s", ex.getMessage()), ex); + return false; } + return true; + } + + @Override + protected boolean deactivate(LifecycleEvent event) { + return cacheService.deactivate(); } @Override diff --git a/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheService.java b/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheService.java index 1cc6b62a1fb..0d25b851f6a 100644 --- a/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheService.java +++ b/src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheService.java @@ -64,6 +64,7 @@ @Slf4j public class KildaEntryCacheService { + private boolean active; private final FlowRepository commonFlowRepository; private final YFlowRepository yFlowRepository; @@ -80,6 +81,7 @@ public KildaEntryCacheService(PersistenceManager persistenceManager, KildaEntryC this.commonFlowRepository = repositoryFactory.createFlowRepository(); this.yFlowRepository = repositoryFactory.createYFlowRepository(); this.carrier = carrier; + this.active = false; } /** @@ -139,12 +141,38 @@ public void removeCached(RemoveYFlowStatsInfo yFlowStatsInfo) { /** * Refresh the cache by rereading the flow data from the persistence. */ - public void refreshCache() { + private void refreshCache() { refreshCommonFlowsCache(); refreshYFlowsCache(); log.debug("cookieToFlow cache: {}, switchAndMeterToFlow cache: {}", cookieToFlow, switchAndMeterToFlow); } + private void clearCache() { + cookieToFlow.clear(); + switchAndMeterToFlow.clear(); + } + + /** + * Activates the service. Fills cache in particular. + */ + public void activate() { + if (!active) { + refreshCache(); + } + active = true; + } + + /** + * Deactivates the service. Clears cache in particular. + */ + public boolean deactivate() { + if (active) { + clearCache(); + } + active = false; + return true; + } + private void updateCache(KildaEntryDescriptorHandler cacheHandler, BaseFlowPathInfo pathInfo) { updateCache( cacheHandler, pathInfo.getFlowId(), pathInfo.getYFlowId(), pathInfo.getCookie(), pathInfo.getMeterId(), diff --git a/src-java/stats-topology/stats-storm-topology/src/test/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheServiceTest.java b/src-java/stats-topology/stats-storm-topology/src/test/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheServiceTest.java index 4a1506362e1..952a3f5566b 100644 --- a/src-java/stats-topology/stats-storm-topology/src/test/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheServiceTest.java +++ b/src-java/stats-topology/stats-storm-topology/src/test/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheServiceTest.java @@ -16,7 +16,10 @@ package org.openkilda.wfm.topology.stats.service; import static java.util.Arrays.asList; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.openkilda.wfm.topology.stats.model.MeasurePoint.EGRESS; @@ -134,7 +137,7 @@ public void shouldRefreshCommonFlowsCookieCache() { when(flowRepository.findAll()).thenReturn(Collections.singletonList(flow)); when(yFlowRepository.findAll()).thenReturn(Collections.emptyList()); - service.refreshCache(); + service.activate(); final FlowPath forwardPath = flow.getForwardPath(); @@ -265,7 +268,7 @@ public void shouldRefreshYFlowSubFlowCookieCache() { when(flowRepository.findAll()).thenReturn(Collections.singletonList(flow)); when(yFlowRepository.findAll()).thenReturn(Collections.emptyList()); - service.refreshCache(); + service.activate(); final FlowSegmentCookie forwardPathCookie = flow.getForwardPath().getCookie(); final FlowSegmentCookie reversePathCookie = flow.getReversePath().getCookie(); @@ -332,7 +335,7 @@ public void shouldCacheServiceRefreshMeterCache() { Flow flow = buildFlow(); when(flowRepository.findAll()).thenReturn(Collections.singletonList(flow)); - service.refreshCache(); + service.activate(); MeterStatsData statsOriginSrc = getMeterStatsDataSrcSwitch(); service.completeAndForwardMeterStats(statsOriginSrc); @@ -423,7 +426,7 @@ public void shouldRefreshYFlowMeterCache() { when(flowRepository.findAll()).thenReturn(Collections.emptyList()); when(yFlowRepository.findAll()).thenReturn(Collections.singletonList(yFlow)); - service.refreshCache(); + service.activate(); // shared endpoint service.completeAndForwardMeterStats(new MeterStatsData( @@ -660,6 +663,73 @@ public void shouldHandleRemovingMeterFromCache() { assertDescriptionPopulation(statsEntries, statsOrigin.getStats().size(), 0); } + @Test + public void serviceActivationAndDeactivationTest() { + Flow flow = buildFlow(); + when(flowRepository.findAll()).thenReturn(Collections.singletonList(flow)); + when(yFlowRepository.findAll()).thenReturn(Collections.emptyList()); + + FlowStatsData flowStats = new FlowStatsData(SRC_SWITCH_ID, Collections.singletonList( + new FlowStatsEntry(0, FORWARD_PATH_COOKIE.getValue(), 0, 0, 0, 0))); + + service.activate(); + service.completeAndForwardFlowStats(flowStats); + verify(carrier, atLeastOnce()).emitFlowStats(cookieCacheCaptor.capture()); + assertEquals(1, cookieCacheCaptor.getValue().getStatsEntries().size()); + assertEquals(flow.getFlowId(), ((CommonFlowDescriptor) cookieCacheCaptor.getValue().getStatsEntries().get(0) + .getDescriptor()).getFlowId()); + + service.deactivate(); + service.completeAndForwardFlowStats(flowStats); + verify(carrier, atLeastOnce()).emitFlowStats(cookieCacheCaptor.capture()); + assertEquals(1, cookieCacheCaptor.getValue().getStatsEntries().size()); + assertNull(cookieCacheCaptor.getValue().getStatsEntries().get(0).getDescriptor()); + + service.activate(); + service.completeAndForwardFlowStats(flowStats); + verify(carrier, atLeastOnce()).emitFlowStats(cookieCacheCaptor.capture()); + assertEquals(1, cookieCacheCaptor.getValue().getStatsEntries().size()); + assertEquals(flow.getFlowId(), ((CommonFlowDescriptor) cookieCacheCaptor.getValue().getStatsEntries().get(0) + .getDescriptor()).getFlowId()); + } + + @Test + public void serviceSingleActivationTest() { + when(flowRepository.findAll()).thenReturn(Collections.emptyList()); + when(yFlowRepository.findAll()).thenReturn(Collections.emptyList()); + + service.activate(); + + verify(flowRepository, times(1)).findAll(); + verify(yFlowRepository, times(1)).findAll(); + } + + @Test + public void serviceDoubleActivationTest() { + when(flowRepository.findAll()).thenReturn(Collections.emptyList()); + when(yFlowRepository.findAll()).thenReturn(Collections.emptyList()); + + service.activate(); + service.activate(); // second activation must not refresh cache + + verify(flowRepository, times(1)).findAll(); + verify(yFlowRepository, times(1)).findAll(); + } + + + @Test + public void serviceActivationAfterDeactivationTest() { + when(flowRepository.findAll()).thenReturn(Collections.emptyList()); + when(yFlowRepository.findAll()).thenReturn(Collections.emptyList()); + + service.activate(); + service.deactivate(); + service.activate(); + + verify(flowRepository, times(2)).findAll(); + verify(yFlowRepository, times(2)).findAll(); + } + private void assertCookieCache( List statsEntries, FlowSegmentCookie cookie, KildaEntryDescriptor expectedDescriptor) { @@ -667,7 +737,7 @@ private void assertCookieCache( for (FlowStatsAndDescriptor entry : statsEntries) { if (needle == entry.getData().getCookie()) { KildaEntryDescriptor descriptor = entry.getDescriptor(); - Assert.assertEquals(expectedDescriptor, descriptor); + assertEquals(expectedDescriptor, descriptor); return; } } @@ -679,7 +749,7 @@ private void assertMeterCache( for (MeterStatsAndDescriptor entry : statsEntries) { if (meterId == entry.getData().getMeterId()) { KildaEntryDescriptor descriptor = entry.getDescriptor(); - Assert.assertEquals(expectedDescriptor, descriptor); + assertEquals(expectedDescriptor, descriptor); return; } } @@ -688,8 +758,8 @@ private void assertMeterCache( private void assertDescriptionPopulation( List> statsEntries, long expectEntriesTotal, long expectCacheHits) { - Assert.assertEquals(expectEntriesTotal, statsEntries.size()); - Assert.assertEquals( + assertEquals(expectEntriesTotal, statsEntries.size()); + assertEquals( expectCacheHits, statsEntries.stream().filter(entry -> entry.getDescriptor() != null).count()); } diff --git a/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/bolt/SwitchManagerHub.java b/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/bolt/SwitchManagerHub.java index b78bd46e911..cea99ddd850 100644 --- a/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/bolt/SwitchManagerHub.java +++ b/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/bolt/SwitchManagerHub.java @@ -23,7 +23,6 @@ import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.ModifySpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.floodlight.api.response.rulemanager.SpeakerCommandResponse; import org.openkilda.messaging.AbstractMessage; import org.openkilda.messaging.Message; @@ -423,7 +422,6 @@ private BaseSpeakerCommandsRequest getRequest(String requestKey, List .switchId(switchId) .commandId(commandId) .commands(commands) - .origin(Origin.SW_MANAGER) .build(); case INSTALL_IF_NOT_EXIST: return InstallSpeakerCommandsRequest.builder() @@ -431,15 +429,12 @@ private BaseSpeakerCommandsRequest getRequest(String requestKey, List .switchId(switchId) .commandId(commandId) .commands(commands) - .origin(Origin.SW_MANAGER) .failIfExists(false) .build(); case MODIFY: - return new ModifySpeakerCommandsRequest(messageContext, switchId, commandId, commands, - Origin.SW_MANAGER); + return new ModifySpeakerCommandsRequest(messageContext, switchId, commandId, commands); case DELETE: - return new DeleteSpeakerCommandsRequest(messageContext, switchId, commandId, commands, - Origin.SW_MANAGER); + return new DeleteSpeakerCommandsRequest(messageContext, switchId, commandId, commands); default: throw new IllegalStateException(format("Unknown OpenFlow command type %s", action)); } diff --git a/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/fsm/DeleteLagPortFsm.java b/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/fsm/DeleteLagPortFsm.java index c8b52b1d386..cb82912eac2 100644 --- a/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/fsm/DeleteLagPortFsm.java +++ b/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/fsm/DeleteLagPortFsm.java @@ -140,7 +140,8 @@ void sendGrpcRequest(DeleteLagState from, DeleteLagState to, DeleteLagEvent even } void removeDbLag(DeleteLagState from, DeleteLagState to, DeleteLagEvent event, DeleteLagContext context) { - log.info("Removing LAG port {} from database. Switch {}. Key={}", context.deletedLogicalPort, switchId, key); + log.info("Removing LAG port {} from database. Switch {}. Key={}", + request.getLogicalPortNumber(), switchId, key); Integer portNumber = request.getLogicalPortNumber(); try { removedLagPort = lagPortOperationService.removeLagPort(switchId, portNumber); diff --git a/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/service/DeleteLagPortService.java b/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/service/DeleteLagPortService.java index 065da10826e..fba6b5659f7 100644 --- a/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/service/DeleteLagPortService.java +++ b/src-java/swmanager-topology/swmanager-storm-topology/src/main/java/org/openkilda/wfm/topology/switchmanager/service/DeleteLagPortService.java @@ -17,6 +17,7 @@ import org.openkilda.messaging.MessageCookie; import org.openkilda.messaging.error.ErrorData; +import org.openkilda.messaging.error.ErrorType; import org.openkilda.messaging.info.InfoData; import org.openkilda.messaging.info.grpc.DeleteLogicalPortResponse; import org.openkilda.messaging.swmanager.request.DeleteLagPortRequest; @@ -92,6 +93,15 @@ public void dispatchErrorMessage(ErrorData payload, MessageCookie cookie) throws DeleteLagContext context = DeleteLagContext.builder() .error(new SpeakerFailureException(payload)) .build(); + + if (payload.getErrorType() == ErrorType.NOT_FOUND) { + DeleteLagPortFsm fsm = fetchHandler(cookie); + log.warn("The port {} is stored in the database but does not exist on the switch {}", + fsm.getRequest().getLogicalPortNumber(), fsm.getSwitchId()); + fireFsmEvent(cookie, DeleteLagEvent.LAG_REMOVED, context); + return; + } + fireFsmEvent(cookie, DeleteLagEvent.ERROR, context); } @@ -119,14 +129,8 @@ public boolean isAllOperationsCompleted() { private void fireFsmEvent(MessageCookie cookie, DeleteLagEvent event, DeleteLagContext context) throws MessageDispatchException { - DeleteLagPortFsm handler = null; - if (cookie != null) { - handler = handlers.get(cookie.getValue()); - } - if (handler == null) { - throw new MessageDispatchException(cookie); - } - fireFsmEvent(handler, event, context); + DeleteLagPortFsm fsm = fetchHandler(cookie); + fireFsmEvent(fsm, event, context); } private void fireFsmEvent(DeleteLagPortFsm fsm, DeleteLagEvent event, DeleteLagContext context) { @@ -146,4 +150,15 @@ private void removeIfCompleted(DeleteLagPortFsm fsm) { } } } + + private DeleteLagPortFsm fetchHandler(MessageCookie cookie) throws MessageDispatchException { + DeleteLagPortFsm handler = null; + if (cookie != null) { + handler = handlers.get(cookie.getValue()); + } + if (handler == null) { + throw new MessageDispatchException(cookie); + } + return handler; + } } diff --git a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/MirrorEndpointsSpec.groovy b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/MirrorEndpointsSpec.groovy index f6d925e05f9..98ebc92cdb0 100644 --- a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/MirrorEndpointsSpec.groovy +++ b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/MirrorEndpointsSpec.groovy @@ -158,12 +158,11 @@ class MirrorEndpointsSpec extends HealthCheckSpecification { and: "Original flow rule counter is not increased" flowRule.packetCount == findFlowRule(getFlowRules(swPair.src.dpId), mirrorDirection).packetCount - //https://github.com/telstra/open-kilda/issues/4517 -// and: "System collects stat for mirror cookie in otsdb" -// Wrappers.wait(statsRouterRequestInterval) { -// def tags = [flowid: flow.flowId, cookie: mirrorRule.cookie] -// assert otsdb.query(genTrafficTime, metricPrefix + "flow.raw.bytes", tags).dps.size() > 0 -// } + and: "System collects stat for mirror cookie in otsdb" + Wrappers.wait(statsRouterRequestInterval) { + def tags = [flowid: flow.flowId, cookie: mirrorRule.cookie] + assert otsdb.query(genTrafficTime, metricPrefix + "flow.raw.bytes", tags).dps.size() > 0 + } and: "Traffic is also received at the mirror point (check only if second tg available)" if (mirrorTg) { diff --git a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/PartialUpdateSpec.groovy b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/PartialUpdateSpec.groovy index fdae5b3f32a..db1e816ffd3 100644 --- a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/PartialUpdateSpec.groovy +++ b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/flows/PartialUpdateSpec.groovy @@ -105,15 +105,6 @@ class PartialUpdateSpec extends HealthCheckSpecification { field : "pinned", newValue: true ], - //https://github.com/telstra/open-kilda/issues/3896 -// [ -// field : "pathComputationStrategy", -// newValue: PathComputationStrategy.LATENCY.toString().toLowerCase() -// ], -// [ -// field : "ignoreBandwidth", -// newValue: true -// ] [ field : "description", newValue: "updated" diff --git a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/network/PathsSpec.groovy b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/network/PathsSpec.groovy index c4a9b7b7156..51ca90a9a59 100644 --- a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/network/PathsSpec.groovy +++ b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/network/PathsSpec.groovy @@ -29,7 +29,8 @@ class PathsSpec extends HealthCheckSpecification { def flow = flowHelperV2.addFlow(flowHelperV2.randomFlow(switchPair)) when: "Get paths between switches" - def paths = northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, null, null, null, null) + def paths = northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, + null, null, null, null, null) then: "Paths will be sorted by bandwidth (descending order) and then by latency (ascending order)" paths.paths.size() > 0 @@ -45,6 +46,13 @@ class PathsSpec extends HealthCheckSpecification { } } + then: "Maximum count of paths can be changed during PCE calculations" + def limited_paths = northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, + null, null, null, null, 1) + assert limited_paths.paths.size() == 1 + assert paths.paths.size() > limited_paths.paths.size() + + cleanup: flow && flowHelperV2.deleteFlow(flow.flowId) } @@ -59,8 +67,8 @@ class PathsSpec extends HealthCheckSpecification { def flow = flowHelperV2.addFlow(flowHelperV2.randomFlow(switchPair)) when: "Get paths between switches using the LATENCY strategy" - def paths = northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, null, - PathComputationStrategy.LATENCY, 10, null) + def paths = northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, + null, PathComputationStrategy.LATENCY, 10, null, null) then: "Paths will be sorted by latency (ascending order) and then by bandwidth (descending order)" paths.paths.size() > 0 @@ -85,7 +93,7 @@ class PathsSpec extends HealthCheckSpecification { def sw = topology.getActiveSwitches()[0] when: "Try to get paths between one switch" - northbound.getPaths(sw.dpId, sw.dpId, null, null, null, null) + northbound.getPaths(sw.dpId, sw.dpId, null, null, null, null, null) then: "Get 400 BadRequest error because request is invalid" def exc = thrown(HttpClientErrorException) @@ -99,7 +107,7 @@ class PathsSpec extends HealthCheckSpecification { def sw = topology.getActiveSwitches()[0] when: "Try to get paths between real switch and nonexistent switch" - northbound.getPaths(sw.dpId, NON_EXISTENT_SWITCH_ID, null, null, null, null) + northbound.getPaths(sw.dpId, NON_EXISTENT_SWITCH_ID, null, null, null, null, null) then: "Get 404 NotFound error" def exc = thrown(HttpClientErrorException) @@ -112,8 +120,8 @@ class PathsSpec extends HealthCheckSpecification { def switchPair = topologyHelper.getNotNeighboringSwitchPair() when: "Try to get paths between switches with max_latency stragy but without max_latency parameter" - northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, null, PathComputationStrategy.MAX_LATENCY, - null, null) + northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, + null, PathComputationStrategy.MAX_LATENCY, null, null, null) then: "Human readable error is returned" def error = thrown(HttpClientErrorException) @@ -142,7 +150,8 @@ class PathsSpec extends HealthCheckSpecification { .supportedTransitEncapsulation.collect { it.toString().toUpperCase() } when: "Try to get a path for a 'vxlan' flowEncapsulationType between the given switches" - northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, FlowEncapsulationType.VXLAN, null, null, null) + northbound.getPaths(switchPair.src.dpId, switchPair.dst.dpId, + FlowEncapsulationType.VXLAN, null, null, null, null) then: "Human readable error is returned" def exc = thrown(HttpClientErrorException) diff --git a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/switches/LagPortSpec.groovy b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/switches/LagPortSpec.groovy index 35a6a28a670..d235ba99649 100644 --- a/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/switches/LagPortSpec.groovy +++ b/src-java/testing/functional-tests/src/test/groovy/org/openkilda/functionaltests/spec/switches/LagPortSpec.groovy @@ -623,6 +623,29 @@ class LagPortSpec extends HealthCheckSpecification { lagPort && northboundV2.deleteLagLogicalPort(switchPair.src.dpId, lagPort) } + @Tidy + def "Able to delete LAG port if it is already removed from switch"() { + given: "A switch with a LAG port" + def sw = topology.getActiveSwitches().first() + def portsArray = topology.getAllowedPortsForSwitch(sw)[-2,-1] + def payload = new LagPortRequest(portNumbers: portsArray) + def lagPort = northboundV2.createLagLogicalPort(sw.dpId, payload).logicalPortNumber + + when: "Delete LAG port via grpc" + grpc.deleteSwitchLogicalPort(northbound.getSwitch(sw.dpId).address, lagPort) + + then: "Able to delete LAG port from switch with no exception" + def deleteResponse = northboundV2.deleteLagLogicalPort(sw.dpId, lagPort) + + with(deleteResponse) { + logicalPortNumber == lagPort + portNumbers.sort() == portsArrayUpdate.sort() + } + + cleanup: + lagPort && northboundV2.deleteLagLogicalPort(sw.dpId, lagPort) + } + void deleteAllLagPorts(SwitchId switchId) { northboundV2.getLagLogicalPort(switchId)*.logicalPortNumber.each { Integer lagPort -> northboundV2.deleteLagLogicalPort(switchId, lagPort) diff --git a/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundService.java b/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundService.java index 844b52fcd66..3e5401b2968 100644 --- a/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundService.java +++ b/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundService.java @@ -209,7 +209,8 @@ LinkMaxBandwidthDto updateLinkMaxBandwidth(SwitchId srcSwitch, Integer srcPort, //feature network PathsDto getPaths(SwitchId srcSwitch, SwitchId dstSwitch, FlowEncapsulationType flowEncapsulationType, - PathComputationStrategy pathComputationStrategy, Long maxLatency, Long maxLatencyTier2); + PathComputationStrategy pathComputationStrategy, Long maxLatency, + Long maxLatencyTier2, Integer maxPathCount); // configuration diff --git a/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundServiceImpl.java b/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundServiceImpl.java index 38da43fd79f..55f3ded6ed5 100644 --- a/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundServiceImpl.java +++ b/src-java/testing/test-library/src/main/java/org/openkilda/testing/service/northbound/NorthboundServiceImpl.java @@ -648,7 +648,8 @@ public List getSwitchFlows(SwitchId switchId, Integer port) { @Override public PathsDto getPaths(SwitchId srcSwitch, SwitchId dstSwitch, FlowEncapsulationType flowEncapsulationType, - PathComputationStrategy pathComputationStrategy, Long maxLatency, Long maxLatencyTier2) { + PathComputationStrategy pathComputationStrategy, + Long maxLatency, Long maxLatencyTier2, Integer maxPathCount) { UriComponentsBuilder uriBuilder = UriComponentsBuilder.fromUriString("/api/v1/network/paths"); if (srcSwitch != null) { uriBuilder.queryParam("src_switch", srcSwitch); @@ -668,6 +669,9 @@ public PathsDto getPaths(SwitchId srcSwitch, SwitchId dstSwitch, FlowEncapsulati if (maxLatencyTier2 != null) { uriBuilder.queryParam("max_latency_tier2", maxLatencyTier2); } + if (maxPathCount != null) { + uriBuilder.queryParam("max_path_count", maxPathCount); + } return restTemplate.exchange(uriBuilder.build().toString(), HttpMethod.GET, new HttpEntity(buildHeadersWithCorrelationId()), PathsDto.class, srcSwitch, dstSwitch).getBody(); diff --git a/src-java/testing/test-library/src/main/java/org/openkilda/testing/tools/KafkaUtils.java b/src-java/testing/test-library/src/main/java/org/openkilda/testing/tools/KafkaUtils.java index 084956c5da4..4e95c27e17f 100644 --- a/src-java/testing/test-library/src/main/java/org/openkilda/testing/tools/KafkaUtils.java +++ b/src-java/testing/test-library/src/main/java/org/openkilda/testing/tools/KafkaUtils.java @@ -22,7 +22,6 @@ import org.openkilda.floodlight.api.request.rulemanager.InstallSpeakerCommandsRequest; import org.openkilda.floodlight.api.request.rulemanager.MeterCommand; import org.openkilda.floodlight.api.request.rulemanager.OfCommand; -import org.openkilda.floodlight.api.request.rulemanager.Origin; import org.openkilda.messaging.AbstractMessage; import org.openkilda.messaging.MessageContext; import org.openkilda.model.FlowPathDirection; @@ -58,7 +57,6 @@ public static AbstractMessage buildMessage(List speakerData) { .switchId(speakerData.get(0).getSwitchId()) .commandId(UUID.randomUUID()) .commands(toCommands(speakerData)) - .origin(Origin.SW_MANAGER) .build(); }