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 race conditions
  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             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'cps-2478-A', cmHandlesPerTag, offset)
  83         and: 'register anther 250 cm handles with module set tag cps-2478-B'
  84             offset += cmHandlesPerTag
  85             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'cps-2478-B', cmHandlesPerTag, offset)
  86         and: 'clear any previous instrumentation'
  87             meterRegistry.clear()
  88         when: 'sync all advised cm handles'
  89             objectUnderTest.moduleSyncAdvisedCmHandles()
  90             Thread.sleep(100)
  91         then: 'Keep processing until there are no more LOCKED or ADVISED cm handles'
  92             new PollingConditions().within(10, () -> {
  93                 def advised = cmHandlesByState.get('advisedCmHandlesCount')
  94                 def locked = cmHandlesByState.get('lockedCmHandlesCount')
  95                 if ( locked > 0 | advised > 0 ) {
  96                     println "CPS-2576 Need to retry ${locked} LOCKED / ${advised} ADVISED cm Handles"
  97                     objectUnderTest.moduleSyncAdvisedCmHandles()
  98                     Thread.sleep(100)
  99                 }
 100                 assert cmHandlesByState.get('lockedCmHandlesCount') + cmHandlesByState.get('advisedCmHandlesCount') == 0
 101             })
 102         and: 'log the relevant instrumentation'
 103             def dmiModuleRetrievalTimer = meterRegistry.get('cps.ncmp.inventory.module.references.from.dmi').timer()
 104             def dbSchemaSetStorageTimer = meterRegistry.get('cps.module.persistence.schemaset.createFromNewAndExistingModules').timer()
 105             def dbStateUpdateTimer = meterRegistry.get('cps.ncmp.cmhandle.state.update.batch').timer()
 106             logInstrumentation(dmiModuleRetrievalTimer, 'get modules from DMI   ')
 107             logInstrumentation(dbSchemaSetStorageTimer, 'store schema sets      ')
 108             logInstrumentation(dbStateUpdateTimer,      'batch state updates    ')
 109         cleanup: 'remove all test cm handles'
 110             // To properly measure performance the sample-size should be increased to 20,000 cm handles or higher (10,000 per tag)
 111             def stopWatch = new StopWatch()
 112             stopWatch.start()
 113             deregisterSequenceOfCmHandles(DMI1_URL, totalCmHandles, 1)
 114             stopWatch.stop()
 115             println "*** CPS-2478, Deletion of $totalCmHandles cm handles took ${stopWatch.getTotalTimeMillis()} milliseconds"
 116     }
 117
 118     def 'Populate module sync work queue simultaneously on two parallel threads (CPS-2403).'() {
 119         // This test failed before bug https://lf-onap.atlassian.net/browse/CPS-2403 was fixed
 120         given: 'the queue is empty at the start'
 121             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, NO_MODULE_SET_TAG, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 122             assert moduleSyncWorkQueue.isEmpty()
 123         when: 'attempt to populate the queue on the main (test) and another parallel thread at the same time'
 124             objectUnderTest.populateWorkQueueIfNeeded()
 125             executorService.execute(populateQueueWithoutDelay)
 126         and: 'wait a little (to give all threads time to complete their task)'
 127             Thread.sleep(50)
 128         then: 'the queue size is exactly the sample size'
 129             assert moduleSyncWorkQueue.size() == PARALLEL_SYNC_SAMPLE_SIZE
 130         cleanup: 'remove all test cm handles'
 131             deregisterSequenceOfCmHandles(DMI1_URL, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 132     }
 133
 134     /** this test has intermittent failures, due to race conditions
 135      *  Ignored but left here as it might be valuable to further optimization investigations.
 136      **/
 137     @Ignore
 138     def 'Schema sets with overlapping modules processed at the same time (DB constraint violation).'() {
 139         given: 'register one batch (100) cm handles of tag A (with overlapping module names)'
 140             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'tagA', 100, 1, ModuleNameStrategy.OVERLAPPING)
 141         and: 'register another batch cm handles of tag B (with overlapping module names)'
 142             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, 'tagB', 100, 101, ModuleNameStrategy.OVERLAPPING)
 143         and: 'populate the work queue with both batches'
 144             objectUnderTest.populateWorkQueueIfNeeded()
 145         when: 'advised cm handles are processed on 2 threads (exactly one batch for each)'
 146             objectUnderTest.moduleSyncAdvisedCmHandles()
 147             executorService.execute(moduleSyncAdvisedCmHandles)
 148         then: 'all cm handles have been processed'
 149             assert getNumberOfProcessedCmHandles() == 200
 150         then: 'at least 1 cm handle is in state LOCKED'
 151             assert cmHandlesByState.get('lockedCmHandlesCount') >= 1
 152         cleanup: 'remove all test cm handles'
 153             deregisterSequenceOfCmHandles(DMI1_URL, 200, 1)
 154     }
 155
 156     def 'Populate module sync work queue on two parallel threads with a slight difference in start time.'() {
 157         // This test proved that the issue in CPS-2403 did not arise if the the queue was populated and given time to be distributed
 158         given: 'the queue is empty at the start'
 159             registerSequenceOfCmHandlesWithManyModuleReferencesButDoNotWaitForReady(DMI1_URL, NO_MODULE_SET_TAG, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 160             assert moduleSyncWorkQueue.isEmpty()
 161         when: 'attempt to populate the queue on the main (test) and another parallel thread a little later'
 162             objectUnderTest.populateWorkQueueIfNeeded()
 163             executorService.execute(populateQueueWithDelay)
 164         and: 'wait a little (to give all threads time to complete their task)'
 165             Thread.sleep(50)
 166         then: 'the queue size is exactly the sample size'
 167             assert moduleSyncWorkQueue.size() == PARALLEL_SYNC_SAMPLE_SIZE
 168         cleanup: 'remove all test cm handles'
 169             deregisterSequenceOfCmHandles(DMI1_URL, PARALLEL_SYNC_SAMPLE_SIZE, 1)
 170     }
 171
 172     def logInstrumentation(timer, description) {
 173         println "*** CPS-2478, $description : Invoked ${timer.count()} times, Total Time: ${timer.totalTime(TimeUnit.MILLISECONDS)} ms, Mean Time: ${timer.mean(TimeUnit.MILLISECONDS)} ms"
 174         return true
 175     }
 176
 177     def populateQueueWithoutDelay = () -> {
 178         try {
 179             objectUnderTest.populateWorkQueueIfNeeded()
 180         } catch (InterruptedException e) {
 181             e.printStackTrace()
 182         }
 183     }
 184
 185     def populateQueueWithDelay = () -> {
 186         try {
 187             Thread.sleep(10)
 188             objectUnderTest.populateWorkQueueIfNeeded()
 189         } catch (InterruptedException e) {
 190             e.printStackTrace()
 191         }
 192     }
 193
 194     def moduleSyncAdvisedCmHandles = () -> {
 195         try {
 196             objectUnderTest.moduleSyncAdvisedCmHandles()
 197         } catch (InterruptedException e) {
 198             e.printStackTrace()
 199         }
 200     }
 201
 202     def clearCmHandleStateGauge() {
 203         cmHandlesByState.keySet().each { cmHandlesByState.put(it, 0)}
 204     }
 205
 206     def getNumberOfProcessedCmHandles() {
 207         return cmHandlesByState.get('readyCmHandlesCount') + cmHandlesByState.get('lockedCmHandlesCount')
 208     }
 209
 210
 211 }