gerrit.onap Code Review - cps.git/blob

   1 /*
   2  *  ============LICENSE_START=======================================================
   3  *  Copyright (C) 2024-2025 Nordix Foundation
   4  *  ================================================================================
   5  *  Licensed under the Apache License, Version 2.0 (the 'License');
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at
   8  *
   9  *        http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an 'AS IS' BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  *
  17  *  SPDX-License-Identifier: Apache-2.0
  18  *  ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.cps.integration.functional.ncmp
  22
  23 import com.hazelcast.map.IMap
  24 import io.micrometer.core.instrument.MeterRegistry
  25 import org.onap.cps.integration.base.CpsIntegrationSpecBase
  26 import org.onap.cps.ncmp.impl.inventory.sync.ModuleSyncWatchdog
  27 import org.springframework.beans.factory.annotation.Autowired
  28 import org.springframework.util.StopWatch
  29 import spock.lang.Ignore
  30 import spock.util.concurrent.PollingConditions
  31
  32 import java.util.concurrent.Executors
  33 import java.util.concurrent.TimeUnit
  34
  35 class ModuleSyncWatchdogIntegrationSpec extends CpsIntegrationSpecBase {
  36
  37     ModuleSyncWatchdog objectUnderTest
  38
  39     @Autowired
  40     MeterRegistry meterRegistry
  41
  42     @Autowired
  43     IMap<String, Integer> cmHandlesByState
  44
  45     def executorService = Executors.newFixedThreadPool(2)
  46     def PARALLEL_SYNC_SAMPLE_SIZE = 100
  47
  48     def setup() {
  49         objectUnderTest = moduleSyncWatchdog
  50         clearCmHandleStateGauge()
  51     }
  52
  53     def cleanup() {
  54         try {
  55             moduleSyncWorkQueue.clear()
  56         } finally {
  57             executorService.shutdownNow()
  58         }
  59     }
  60
  61     def 'Watchdog is disabled for test.'() {
  62         given: 'some cm handles are registered'
  63             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, NO_MODULE_SET_TAG, PARALLEL_SYNC_SAMPLE_SIZE, 1)
  64         when: 'wait a while but less then the initial delay of 10 minutes'
  65             Thread.sleep(3000)
  66         then: 'the work queue remains empty'
  67             assert moduleSyncWorkQueue.isEmpty()
  68         cleanup: 'remove advised cm handles'
  69             deregisterSequenceOfCmHandles(DMI1_URL, PARALLEL_SYNC_SAMPLE_SIZE, 1)
  70     }
  71
  72     /** this test has intermittent failures, due to timeouts.
  73      *  Ignored but left here as it might be valuable to further optimization investigations.
  74      **/
  75     @Ignore
  76     def 'CPS-2478 Highlight (and improve) module sync inefficiencies.'() {
  77         given: 'register 250 cm handles with module set tag cps-2478-A'
  78             def numberOfTags = 2
  79             def cmHandlesPerTag = 250
  80             def totalCmHandles = numberOfTags * cmHandlesPerTag
  81             def offset = 1
  82             def minimumBatches = totalCmHandles / 100
  83             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'cps-2478-A', cmHandlesPerTag, offset)
  84         and: 'register anther 250 cm handles with module set tag cps-2478-B'
  85             offset += cmHandlesPerTag
  86             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'cps-2478-B', cmHandlesPerTag, offset)
  87         and: 'clear any previous instrumentation'
  88             meterRegistry.clear()
  89         when: 'sync all advised cm handles'
  90             objectUnderTest.moduleSyncAdvisedCmHandles()
  91             Thread.sleep(100)
  92         then: 'retry until both schema sets are stored in db (1 schema set for each module set tag)'
  93             def dbSchemaSetStorageTimer = meterRegistry.get('cps.module.persistence.schemaset.store').timer()
  94             new PollingConditions().within(10, () -> {
  95                 objectUnderTest.moduleSyncAdvisedCmHandles()
  96                 Thread.sleep(100)
  97                 assert dbSchemaSetStorageTimer.count() == 2
  98             })
  99         then: 'wait till at least 5 batches of state updates are done (often more because of retries of locked cm handles)'
 100             def dbStateUpdateTimer = meterRegistry.get('cps.ncmp.cmhandle.state.update.batch').timer()
 101             new PollingConditions().within(10, () -> {
 102                 assert dbStateUpdateTimer.count() >= minimumBatches
 103             })
 104         and: 'one call to DMI per module set tag to get module references (may be more due to parallel processing of batches)'
 105             def dmiModuleRetrievalTimer = meterRegistry.get('cps.ncmp.inventory.module.references.from.dmi').timer()
 106             assert dmiModuleRetrievalTimer.count() >= numberOfTags && dmiModuleRetrievalTimer.count() <= minimumBatches
 107
 108         and: 'log the relevant instrumentation'
 109             logInstrumentation(dmiModuleRetrievalTimer, 'get modules from DMI   ')
 110             logInstrumentation(dbSchemaSetStorageTimer, 'store schema sets      ')
 111             logInstrumentation(dbStateUpdateTimer,      'batch state updates    ')
 112         cleanup: 'remove all test cm handles'
 113             // To properly measure performance the sample-size should be increased to 20,000 cm handles or higher (10,000 per tag)
 114             def stopWatch = new StopWatch()
 115             stopWatch.start()
 116             deregisterSequenceOfCmHandles(DMI1_URL, totalCmHandles, 1)
 117             stopWatch.stop()
 118             println "*** CPS-2478, Deletion of $totalCmHandles cm handles took ${stopWatch.getTotalTimeMillis()} milliseconds"
 119     }
 120
 121     def 'Populate module sync work queue simultaneously on two parallel threads (CPS-2403).'() {
 122         // This test failed before bug https://lf-onap.atlassian.net/browse/CPS-2403 was fixed
 123         given: 'the queue is empty at the start'
 124             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, NO_MODULE_SET_TAG, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 125             assert moduleSyncWorkQueue.isEmpty()
 126         when: 'attempt to populate the queue on the main (test) and another parallel thread at the same time'
 127             objectUnderTest.populateWorkQueueIfNeeded()
 128             executorService.execute(populateQueueWithoutDelay)
 129         and: 'wait a little (to give all threads time to complete their task)'
 130             Thread.sleep(50)
 131         then: 'the queue size is exactly the sample size'
 132             assert moduleSyncWorkQueue.size() == PARALLEL_SYNC_SAMPLE_SIZE
 133         cleanup: 'remove all test cm handles'
 134             deregisterSequenceOfCmHandles(DMI1_URL, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 135     }
 136
 137     def 'Schema sets with overlapping modules processed at the same time (DB constraint violation).'() {
 138         given: 'register one batch (100) cm handles of tag A (with overlapping module names)'
 139             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'tagA', 100, 1, ModuleNameStrategy.OVERLAPPING)
 140         and: 'register another batch cm handles of tag B (with overlapping module names)'
 141             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'tagB', 100, 101, ModuleNameStrategy.OVERLAPPING)
 142         and: 'populate the work queue with both batches'
 143             objectUnderTest.populateWorkQueueIfNeeded()
 144         when: 'advised cm handles are processed on 2 threads (exactly one batch for each)'
 145             objectUnderTest.moduleSyncAdvisedCmHandles()
 146             executorService.execute(moduleSyncAdvisedCmHandles)
 147         then: 'wait till all cm handles have been processed'
 148             new PollingConditions().within(10, () -> {
 149                 assert getNumberOfProcessedCmHandles() == 200
 150             })
 151         then: 'at least 1 cm handle is in state LOCKED'
 152             assert cmHandlesByState.get('lockedCmHandlesCount') >= 1
 153         cleanup: 'remove all test cm handles'
 154             deregisterSequenceOfCmHandles(DMI1_URL, 200, 1)
 155     }
 156
 157     def 'Populate module sync work queue on two parallel threads with a slight difference in start time.'() {
 158         // This test proved that the issue in CPS-2403 did not arise if the the queue was populated and given time to be distributed
 159         given: 'the queue is empty at the start'
 160             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, NO_MODULE_SET_TAG, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 161             assert moduleSyncWorkQueue.isEmpty()
 162         when: 'attempt to populate the queue on the main (test) and another parallel thread a little later'
 163             objectUnderTest.populateWorkQueueIfNeeded()
 164             executorService.execute(populateQueueWithDelay)
 165         and: 'wait a little (to give all threads time to complete their task)'
 166             Thread.sleep(50)
 167         then: 'the queue size is exactly the sample size'
 168             assert moduleSyncWorkQueue.size() == PARALLEL_SYNC_SAMPLE_SIZE
 169         cleanup: 'remove all test cm handles'
 170             deregisterSequenceOfCmHandles(DMI1_URL, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 171     }
 172
 173     def logInstrumentation(timer, description) {
 174         println "*** CPS-2478, $description : Invoked ${timer.count()} times, Total Time: ${timer.totalTime(TimeUnit.MILLISECONDS)} ms, Mean Time: ${timer.mean(TimeUnit.MILLISECONDS)} ms"
 175         return true
 176     }
 177
 178     def populateQueueWithoutDelay = () -> {
 179         try {
 180             objectUnderTest.populateWorkQueueIfNeeded()
 181         } catch (InterruptedException e) {
 182             e.printStackTrace()
 183         }
 184     }
 185
 186     def populateQueueWithDelay = () -> {
 187         try {
 188             Thread.sleep(10)
 189             objectUnderTest.populateWorkQueueIfNeeded()
 190         } catch (InterruptedException e) {
 191             e.printStackTrace()
 192         }
 193     }
 194
 195     def moduleSyncAdvisedCmHandles = () -> {
 196         try {
 197             objectUnderTest.moduleSyncAdvisedCmHandles()
 198         } catch (InterruptedException e) {
 199             e.printStackTrace()
 200         }
 201     }
 202
 203     def clearCmHandleStateGauge() {
 204         cmHandlesByState.keySet().each { cmHandlesByState.put(it, 0)}
 205     }
 206
 207     def getNumberOfProcessedCmHandles() {
 208         return cmHandlesByState.get('readyCmHandlesCount') + cmHandlesByState.get('lockedCmHandlesCount')
 209     }
 210
 211
 212 }