Subsfailure case internal delete and retry in duplicate cause case
[ric-plt/submgr.git] / pkg / control / registry.go
1 /*
2 ==================================================================================
3   Copyright (c) 2019 AT&T Intellectual Property.
4   Copyright (c) 2019 Nokia
5
6    Licensed under the Apache License, Version 2.0 (the "License");
7    you may not use this file except in compliance with the License.
8    You may obtain a copy of the License at
9
10        http://www.apache.org/licenses/LICENSE-2.0
11
12    Unless required by applicable law or agreed to in writing, software
13    distributed under the License is distributed on an "AS IS" BASIS,
14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15    See the License for the specific language governing permissions and
16    limitations under the License.
17 ==================================================================================
18 */
19
20 package control
21
22 import (
23         "fmt"
24         "gerrit.o-ran-sc.org/r/ric-plt/e2ap/pkg/e2ap"
25         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models"
26         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
27         "sync"
28         "time"
29 )
30
31 //-----------------------------------------------------------------------------
32 //
33 //-----------------------------------------------------------------------------
34
35 type Registry struct {
36         mutex       sync.Mutex
37         register    map[uint32]*Subscription
38         subIds      []uint32
39         rtmgrClient *RtmgrClient
40 }
41
42 func (r *Registry) Initialize() {
43         r.register = make(map[uint32]*Subscription)
44         var i uint32
45         for i = 0; i < 65535; i++ {
46                 r.subIds = append(r.subIds, i+1)
47         }
48 }
49
50 func (r *Registry) QueryHandler() (models.SubscriptionList, error) {
51         r.mutex.Lock()
52         defer r.mutex.Unlock()
53
54         resp := models.SubscriptionList{}
55         for _, subs := range r.register {
56                 subs.mutex.Lock()
57                 resp = append(resp, &models.SubscriptionData{SubscriptionID: int64(subs.ReqId.Seq), Meid: subs.Meid.RanName, Endpoint: subs.EpList.StringList()})
58                 subs.mutex.Unlock()
59         }
60         return resp, nil
61 }
62
63 func (r *Registry) allocateSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) {
64         if len(r.subIds) > 0 {
65                 sequenceNumber := r.subIds[0]
66                 r.subIds = r.subIds[1:]
67                 if _, ok := r.register[sequenceNumber]; ok == true {
68                         r.subIds = append(r.subIds, sequenceNumber)
69                         return nil, fmt.Errorf("Registry: Failed to reserve subscription exists")
70                 }
71                 subs := &Subscription{
72                         registry:  r,
73                         Meid:      trans.Meid,
74                         SubReqMsg: subReqMsg,
75                         valid:     true,
76                 }
77                 subs.ReqId.Id = 123
78                 subs.ReqId.Seq = sequenceNumber
79
80                 if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false {
81                         r.subIds = append(r.subIds, subs.ReqId.Seq)
82                         return nil, fmt.Errorf("Registry: Endpoint existing already in subscription")
83                 }
84
85                 return subs, nil
86         }
87         return nil, fmt.Errorf("Registry: Failed to reserve subscription no free ids")
88 }
89
90 func (r *Registry) findExistingSubs(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) *Subscription {
91
92         for _, subs := range r.register {
93                 if subs.IsMergeable(trans, subReqMsg) {
94
95                         //
96                         // check if there has been race conditions
97                         //
98                         subs.mutex.Lock()
99                         //subs has been set to invalid
100                         if subs.valid == false {
101                                 subs.mutex.Unlock()
102                                 continue
103                         }
104                         // If size is zero, entry is to be deleted
105                         if subs.EpList.Size() == 0 {
106                                 subs.mutex.Unlock()
107                                 continue
108                         }
109                         // try to add to endpointlist.
110                         if subs.EpList.AddEndpoint(trans.GetEndpoint()) == false {
111                                 subs.mutex.Unlock()
112                                 continue
113                         }
114                         subs.mutex.Unlock()
115
116                         xapp.Logger.Debug("Registry: Mergeable subs found %s for %s", subs.String(), trans.String())
117                         return subs
118                 }
119         }
120         return nil
121 }
122
123 func (r *Registry) AssignToSubscription(trans *TransactionXapp, subReqMsg *e2ap.E2APSubscriptionRequest) (*Subscription, error) {
124         var err error
125         var newAlloc bool
126         r.mutex.Lock()
127         defer r.mutex.Unlock()
128
129         //
130         // Check validity of subscription action types
131         //
132         actionType, err := r.CheckActionTypes(subReqMsg)
133         if err != nil {
134                 xapp.Logger.Debug("CREATE %s", err)
135                 return nil, err
136         }
137
138         //
139         // Find possible existing Policy subscription
140         //
141         if actionType == e2ap.E2AP_ActionTypePolicy {
142                 if subs, ok := r.register[trans.GetSubId()]; ok {
143                         xapp.Logger.Debug("CREATE %s. Existing subscription for Policy found", subs.String())
144                         subs.SetCachedResponse(nil, true)
145                         return subs, nil
146                 }
147         }
148
149         subs := r.findExistingSubs(trans, subReqMsg)
150         if subs == nil {
151                 subs, err = r.allocateSubs(trans, subReqMsg)
152                 if err != nil {
153                         return nil, err
154                 }
155                 newAlloc = true
156         }
157
158         //
159         // Add to subscription
160         //
161         subs.mutex.Lock()
162         defer subs.mutex.Unlock()
163
164         epamount := subs.EpList.Size()
165
166         r.mutex.Unlock()
167         //
168         // Subscription route updates
169         //
170         if epamount == 1 {
171                 subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.Seq)}
172                 err = r.rtmgrClient.SubscriptionRequestCreate(subRouteAction)
173         } else {
174                 subRouteAction := SubRouteInfo{subs.EpList, uint16(subs.ReqId.Seq)}
175                 err = r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
176         }
177         r.mutex.Lock()
178
179         if err != nil {
180                 if newAlloc {
181                         r.subIds = append(r.subIds, subs.ReqId.Seq)
182                 }
183                 return nil, err
184         }
185
186         if newAlloc {
187                 r.register[subs.ReqId.Seq] = subs
188         }
189         xapp.Logger.Debug("CREATE %s", subs.String())
190         xapp.Logger.Debug("Registry: substable=%v", r.register)
191         return subs, nil
192 }
193
194 func (r *Registry) CheckActionTypes(subReqMsg *e2ap.E2APSubscriptionRequest) (uint64, error) {
195         var reportFound bool = false
196         var policyFound bool = false
197
198         for _, acts := range subReqMsg.ActionSetups {
199                 if acts.ActionType == e2ap.E2AP_ActionTypeReport {
200                         reportFound = true
201                 }
202                 if acts.ActionType == e2ap.E2AP_ActionTypePolicy {
203                         policyFound = true
204                 }
205         }
206         if reportFound == true && policyFound == true {
207                 return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Report and Policy in same RICactions-ToBeSetup-List")
208         }
209         if reportFound == true {
210                 return e2ap.E2AP_ActionTypeReport, nil
211         }
212         if policyFound == true {
213                 return e2ap.E2AP_ActionTypePolicy, nil
214         }
215         return e2ap.E2AP_ActionTypeInvalid, fmt.Errorf("Invalid action type in RICactions-ToBeSetup-List")
216 }
217
218 // TODO: Works with concurrent calls, but check if can be improved
219 func (r *Registry) RemoveFromSubscription(subs *Subscription, trans *TransactionXapp, waitRouteClean time.Duration) error {
220
221         r.mutex.Lock()
222         defer r.mutex.Unlock()
223         subs.mutex.Lock()
224         defer subs.mutex.Unlock()
225
226         delStatus := subs.EpList.DelEndpoint(trans.GetEndpoint())
227         epamount := subs.EpList.Size()
228         seqId := subs.ReqId.Seq
229
230         if delStatus == false {
231                 return nil
232         }
233
234         go func() {
235                 if waitRouteClean > 0 {
236                         time.Sleep(waitRouteClean)
237                 }
238
239                 subs.mutex.Lock()
240                 defer subs.mutex.Unlock()
241                 xapp.Logger.Info("CLEAN %s", subs.String())
242
243                 if epamount == 0 {
244                         //
245                         // Subscription route delete
246                         //
247                         tmpList := RmrEndpointList{}
248                         tmpList.AddEndpoint(trans.GetEndpoint())
249                         subRouteAction := SubRouteInfo{tmpList, uint16(seqId)}
250                         r.rtmgrClient.SubscriptionRequestDelete(subRouteAction)
251
252                         //
253                         // Subscription release
254                         //
255                         r.mutex.Lock()
256                         defer r.mutex.Unlock()
257
258                         if _, ok := r.register[seqId]; ok {
259                                 xapp.Logger.Debug("RELEASE %s", subs.String())
260                                 delete(r.register, seqId)
261                                 xapp.Logger.Debug("Registry: substable=%v", r.register)
262                         }
263                         r.subIds = append(r.subIds, seqId)
264
265                 } else if subs.EpList.Size() > 0 {
266                         //
267                         // Subscription route updates
268                         //
269                         subRouteAction := SubRouteInfo{subs.EpList, uint16(seqId)}
270                         r.rtmgrClient.SubscriptionRequestUpdate(subRouteAction)
271                 }
272
273         }()
274
275         return nil
276 }
277
278 func (r *Registry) GetSubscription(sn uint32) *Subscription {
279         r.mutex.Lock()
280         defer r.mutex.Unlock()
281         if _, ok := r.register[sn]; ok {
282                 return r.register[sn]
283         }
284         return nil
285 }
286
287 func (r *Registry) GetSubscriptionFirstMatch(ids []uint32) (*Subscription, error) {
288         r.mutex.Lock()
289         defer r.mutex.Unlock()
290         for _, id := range ids {
291                 if _, ok := r.register[id]; ok {
292                         return r.register[id], nil
293                 }
294         }
295         return nil, fmt.Errorf("No valid subscription found with ids %v", ids)
296 }