@@ -66,6 +66,77 @@ func (cm *Manager) CheckWorkflowExistence() {
6666 }
6767}
6868
69+ func getUpgradedKey (wf * wfv1.Workflow , key string , level SyncLevelType ) string {
70+ if wfv1 .CheckHolderKeyVersion (key ) == wfv1 .HoldingNameV1 {
71+ if level == WorkflowLevel {
72+ return getHolderKey (wf , "" )
73+ }
74+ return getHolderKey (wf , key )
75+ }
76+ return key
77+ }
78+
79+ type SyncLevelType int
80+
81+ const (
82+ WorkflowLevel SyncLevelType = 1
83+ TemplateLevel SyncLevelType = 2
84+ ErrorLevel SyncLevelType = 3
85+ )
86+
87+ // HoldingNameV1 keys can be of the form
88+ // x where x is a workflow name
89+ // unfortunately this doesn't differentiate between workflow level keys
90+ // and template level keys. So upgrading is a bit tricky here.
91+
92+ // given a legacy holding name x, namespace y and workflow name z.
93+ // in the case of a workflow level
94+ // if x != z
95+ // upgradedKey := y/z
96+ // elseif x == z
97+ // upgradedKey := y/z
98+ // in the case of a template level
99+ // if x != z
100+ // upgradedKey := y/z/x
101+ // elif x == z
102+ // upgradedKey := y/z/x
103+
104+ // there is a possibility that
105+ // a synchronization exists both at the template level
106+ // and at the workflow level -> impossible to upgrade correctly
107+ // due to ambiguity. Currently we just assume workflow level.
108+ func getWorkflowSyncLevelByName (wf * wfv1.Workflow , lockName string ) (SyncLevelType , error ) {
109+ if wf .Spec .Synchronization != nil {
110+ syncLockName , err := GetLockName (wf .Spec .Synchronization , wf .Namespace )
111+ if err != nil {
112+ return ErrorLevel , err
113+ }
114+ checkName := syncLockName .EncodeName ()
115+ if lockName == checkName {
116+ return WorkflowLevel , nil
117+ }
118+ }
119+
120+ var lastErr error
121+ for _ , template := range wf .Spec .Templates {
122+ if template .Synchronization != nil {
123+ syncLockName , err := GetLockName (template .Synchronization , wf .Namespace )
124+ if err != nil {
125+ lastErr = err
126+ continue
127+ }
128+ checkName := syncLockName .EncodeName ()
129+ if lockName == checkName {
130+ return TemplateLevel , nil
131+ }
132+ }
133+ }
134+ if lastErr == nil {
135+ lastErr = fmt .Errorf ("was unable to determine level for %s" , lockName )
136+ }
137+ return ErrorLevel , lastErr
138+ }
139+
69140func (cm * Manager ) Initialize (wfs []wfv1.Workflow ) {
70141 for _ , wf := range wfs {
71142 if wf .Status .Synchronization == nil {
@@ -86,11 +157,17 @@ func (cm *Manager) Initialize(wfs []wfv1.Workflow) {
86157 }
87158
88159 for _ , holders := range holding .Holders {
89- resourceKey := getResourceKey (wf .Namespace , wf .Name , holders )
90- if semaphore != nil && semaphore .acquire (resourceKey ) {
91- log .Infof ("Lock acquired by %s from %s" , resourceKey , holding .Semaphore )
160+ level , err := getWorkflowSyncLevelByName (& wf , holding .Semaphore )
161+ if err != nil {
162+ log .Warnf ("cannot obtain lock level for '%s' : %v" , holding .Semaphore , err )
163+ continue
164+ }
165+ key := getUpgradedKey (& wf , holders , level )
166+ if semaphore != nil && semaphore .acquire (key ) {
167+ log .Infof ("Lock acquired by %s from %s" , key , holding .Semaphore )
92168 }
93169 }
170+
94171 }
95172 }
96173
@@ -101,8 +178,13 @@ func (cm *Manager) Initialize(wfs []wfv1.Workflow) {
101178 if mutex == nil {
102179 mutex := cm .initializeMutex (holding .Mutex )
103180 if holding .Holder != "" {
104- resourceKey := getResourceKey (wf .Namespace , wf .Name , holding .Holder )
105- mutex .acquire (resourceKey )
181+ level , err := getWorkflowSyncLevelByName (& wf , holding .Mutex )
182+ if err != nil {
183+ log .Warnf ("cannot obtain lock level for '%s' : %v" , holding .Mutex , err )
184+ continue
185+ }
186+ key := getUpgradedKey (& wf , holding .Holder , level )
187+ mutex .acquire (key )
106188 }
107189 cm .syncLockMap [holding .Mutex ] = mutex
108190 }
@@ -214,10 +296,9 @@ func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool {
214296 }
215297
216298 for _ , holderKey := range holding .Holders {
217- resourceKey := getResourceKey (wf .Namespace , wf .Name , holderKey )
218- syncLockHolder .release (resourceKey )
299+ syncLockHolder .release (holderKey )
219300 wf .Status .Synchronization .Semaphore .LockReleased (holderKey , holding .Semaphore )
220- log .Infof ("%s released a lock from %s" , resourceKey , holding .Semaphore )
301+ log .Infof ("%s released a lock from %s" , holderKey , holding .Semaphore )
221302 }
222303 }
223304
@@ -227,8 +308,8 @@ func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool {
227308 if syncLockHolder == nil {
228309 continue
229310 }
230- resourceKey := getResourceKey (wf . Namespace , wf . Name , wf . Name )
231- syncLockHolder .removeFromQueue (resourceKey )
311+ key := getHolderKey (wf , "" )
312+ syncLockHolder .removeFromQueue (key )
232313 }
233314 wf .Status .Synchronization .Semaphore = nil
234315 }
@@ -240,10 +321,9 @@ func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool {
240321 continue
241322 }
242323
243- resourceKey := getResourceKey (wf .Namespace , wf .Name , holding .Holder )
244- syncLockHolder .release (resourceKey )
324+ syncLockHolder .release (holding .Holder )
245325 wf .Status .Synchronization .Mutex .LockReleased (holding .Holder , holding .Mutex )
246- log .Infof ("%s released a lock from %s" , resourceKey , holding .Mutex )
326+ log .Infof ("%s released a lock from %s" , holding . Holder , holding .Mutex )
247327 }
248328
249329 // Remove the pending Workflow level mutex keys
@@ -252,8 +332,8 @@ func (cm *Manager) ReleaseAll(wf *wfv1.Workflow) bool {
252332 if syncLockHolder == nil {
253333 continue
254334 }
255- resourceKey := getResourceKey (wf . Namespace , wf . Name , wf . Name )
256- syncLockHolder .removeFromQueue (resourceKey )
335+ key := getHolderKey (wf , "" )
336+ syncLockHolder .removeFromQueue (key )
257337 }
258338 wf .Status .Synchronization .Mutex = nil
259339 }
@@ -296,14 +376,6 @@ func getHolderKey(wf *wfv1.Workflow, nodeName string) string {
296376 return key
297377}
298378
299- func getResourceKey (namespace , wfName , resourceName string ) string {
300- resourceKey := fmt .Sprintf ("%s/%s" , namespace , wfName )
301- if resourceName != wfName {
302- resourceKey = fmt .Sprintf ("%s/%s" , resourceKey , resourceName )
303- }
304- return resourceKey
305- }
306-
307379func (cm * Manager ) getCurrentLockHolders (lockName string ) []string {
308380 if concurrency , ok := cm .syncLockMap [lockName ]; ok {
309381 return concurrency .getCurrentHolders ()
0 commit comments