capifcore/internal/publishservice/publishservice.go

   1 // -
   2 //   ========================LICENSE_START=================================
   3 //   O-RAN-SC
   4 //   %%
   5 //   Copyright (C) 2022-2023: Nordix Foundation
   6 //   Copyright (C) 2024: OpenInfra Foundation Europe
   7 //   %%
   8 //   Licensed under the Apache License, Version 2.0 (the "License");
   9 //   you may not use this file except in compliance with the License.
  10 //   You may obtain a copy of the License at
  11 //
  12 //        http://www.apache.org/licenses/LICENSE-2.0
  13 //
  14 //   Unless required by applicable law or agreed to in writing, software
  15 //   distributed under the License is distributed on an "AS IS" BASIS,
  16 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17 //   See the License for the specific language governing permissions and
  18 //   limitations under the License.
  19 //   ========================LICENSE_END===================================
  20 //
  21
  22 package publishservice
  23
  24 import (
  25         "fmt"
  26         "net/http"
  27         "path"
  28         "strings"
  29         "sync"
  30
  31         echo "github.com/labstack/echo/v4"
  32         "k8s.io/utils/strings/slices"
  33
  34         "oransc.org/nonrtric/capifcore/internal/common29122"
  35         "oransc.org/nonrtric/capifcore/internal/eventsapi"
  36         publishapi "oransc.org/nonrtric/capifcore/internal/publishserviceapi"
  37
  38         "oransc.org/nonrtric/capifcore/internal/helmmanagement"
  39         "oransc.org/nonrtric/capifcore/internal/providermanagement"
  40
  41         log "github.com/sirupsen/logrus"
  42 )
  43
  44 //go:generate mockery --name PublishRegister
  45 type PublishRegister interface {
  46         // Checks if the provided API is published.
  47         // Returns true if the provided API has been published, false otherwise.
  48         IsAPIPublished(aefId, path string) bool
  49         // Gets all published APIs.
  50         // Returns a list of all APIs that has been published.
  51         GetAllPublishedServices() []publishapi.ServiceAPIDescription
  52 }
  53
  54 type PublishService struct {
  55         publishedServices map[string][]publishapi.ServiceAPIDescription
  56         serviceRegister   providermanagement.ServiceRegister
  57         helmManager       helmmanagement.HelmManager
  58         eventChannel      chan<- eventsapi.EventNotification
  59         lock              sync.Mutex
  60 }
  61
  62 // Creates a service that implements both the PublishRegister and the publishserviceapi.ServerInterface interfaces.
  63 func NewPublishService(serviceRegister providermanagement.ServiceRegister, hm helmmanagement.HelmManager, eventChannel chan<- eventsapi.EventNotification) *PublishService {
  64         return &PublishService{
  65                 helmManager:       hm,
  66                 publishedServices: make(map[string][]publishapi.ServiceAPIDescription),
  67                 serviceRegister:   serviceRegister,
  68                 eventChannel:      eventChannel,
  69         }
  70 }
  71
  72 func (ps *PublishService) getAllAefIds() []string {
  73         ps.lock.Lock()
  74         defer ps.lock.Unlock()
  75
  76         allIds := []string{}
  77         for _, descriptions := range ps.publishedServices {
  78                 for _, description := range descriptions {
  79                         allIds = append(allIds, description.GetAefIds()...)
  80                 }
  81         }
  82         return allIds
  83 }
  84
  85 func (ps *PublishService) IsAPIPublished(aefId, path string) bool {
  86         return slices.Contains(ps.getAllAefIds(), aefId)
  87 }
  88
  89 func (ps *PublishService) GetAllPublishedServices() []publishapi.ServiceAPIDescription {
  90         publishedDescriptions := []publishapi.ServiceAPIDescription{}
  91         for _, descriptions := range ps.publishedServices {
  92                 publishedDescriptions = append(publishedDescriptions, descriptions...)
  93         }
  94         return publishedDescriptions
  95 }
  96
  97 // Retrieve all published APIs.
  98 func (ps *PublishService) GetApfIdServiceApis(ctx echo.Context, apfId string) error {
  99         if !ps.serviceRegister.IsPublishingFunctionRegistered(apfId) {
 100                 errorMsg := fmt.Sprintf("Unable to get the service due to %s api is only available for publishers", apfId)
 101                 return sendCoreError(ctx, http.StatusNotFound, errorMsg)
 102         }
 103
 104         serviceDescriptions := ps.publishedServices[apfId]
 105         err := ctx.JSON(http.StatusOK, serviceDescriptions)
 106         if err != nil {
 107                 // Something really bad happened, tell Echo that our handler failed
 108                 return err
 109         }
 110         return nil
 111 }
 112
 113 // Publish a new API.
 114 func (ps *PublishService) PostApfIdServiceApis(ctx echo.Context, apfId string) error {
 115         var newServiceAPIDescription publishapi.ServiceAPIDescription
 116         errorMsg := "Unable to publish the service due to %s "
 117         err := ctx.Bind(&newServiceAPIDescription)
 118         if err != nil {
 119                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errorMsg, "invalid format for service "+apfId))
 120         }
 121
 122         if !ps.serviceRegister.IsPublishingFunctionRegistered(apfId) {
 123                 return sendCoreError(ctx, http.StatusForbidden, fmt.Sprintf(errorMsg, "api is only available for publishers "+apfId))
 124         }
 125
 126         if err := ps.isServicePublished(newServiceAPIDescription); err != nil {
 127                 return sendCoreError(ctx, http.StatusForbidden, fmt.Sprintf(errorMsg, err))
 128         }
 129
 130         if err := newServiceAPIDescription.Validate(); err != nil {
 131                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errorMsg, err))
 132         }
 133         ps.lock.Lock()
 134         defer ps.lock.Unlock()
 135
 136         registeredFuncs := ps.serviceRegister.GetAefsForPublisher(apfId)
 137         for _, profile := range *newServiceAPIDescription.AefProfiles {
 138                 if !slices.Contains(registeredFuncs, profile.AefId) {
 139                         return sendCoreError(ctx, http.StatusNotFound, fmt.Sprintf(errorMsg, fmt.Sprintf("function %s not registered", profile.AefId)))
 140                 }
 141         }
 142
 143         newServiceAPIDescription.PrepareNewService()
 144
 145         shouldReturn, returnValue := ps.installHelmChart(newServiceAPIDescription, ctx)
 146         if shouldReturn {
 147                 return returnValue
 148         }
 149         go ps.sendEvent(newServiceAPIDescription, eventsapi.CAPIFEventSERVICEAPIAVAILABLE)
 150
 151         _, ok := ps.publishedServices[apfId]
 152         if ok {
 153                 ps.publishedServices[apfId] = append(ps.publishedServices[apfId], newServiceAPIDescription)
 154         } else {
 155                 ps.publishedServices[apfId] = append([]publishapi.ServiceAPIDescription{}, newServiceAPIDescription)
 156         }
 157
 158         uri := ctx.Request().Host + ctx.Request().URL.String()
 159         ctx.Response().Header().Set(echo.HeaderLocation, ctx.Scheme()+`://`+path.Join(uri, *newServiceAPIDescription.ApiId))
 160         err = ctx.JSON(http.StatusCreated, newServiceAPIDescription)
 161         if err != nil {
 162                 // Something really bad happened, tell Echo that our handler failed
 163                 return err
 164         }
 165
 166         return nil
 167 }
 168
 169 func (ps *PublishService) isServicePublished(newService publishapi.ServiceAPIDescription) error {
 170         for _, services := range ps.publishedServices {
 171                 for _, service := range services {
 172                         if err := service.ValidateAlreadyPublished(newService); err != nil {
 173                                 return err
 174                         }
 175                 }
 176         }
 177         return nil
 178 }
 179
 180 func (ps *PublishService) installHelmChart(newServiceAPIDescription publishapi.ServiceAPIDescription, ctx echo.Context) (bool, error) {
 181         info := strings.Split(*newServiceAPIDescription.Description, ",")
 182         if len(info) == 5 {
 183                 err := ps.helmManager.InstallHelmChart(info[1], info[2], info[3], info[4])
 184                 if err != nil {
 185                         return true, sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf("Unable to install Helm chart %s due to: %s", info[3], err.Error()))
 186                 }
 187                 log.Debug("Installed service: ", newServiceAPIDescription.ApiId)
 188         }
 189         return false, nil
 190 }
 191
 192 // Unpublish a published service API.
 193 func (ps *PublishService) DeleteApfIdServiceApisServiceApiId(ctx echo.Context, apfId string, serviceApiId string) error {
 194         serviceDescriptions, ok := ps.publishedServices[string(apfId)]
 195         if ok {
 196                 pos, description := getServiceDescription(serviceApiId, serviceDescriptions)
 197                 if description != nil {
 198                         info := strings.Split(*description.Description, ",")
 199                         if len(info) == 5 {
 200                                 ps.helmManager.UninstallHelmChart(info[1], info[3])
 201                                 log.Debug("Deleted service: ", serviceApiId)
 202                         }
 203                         ps.lock.Lock()
 204                         ps.publishedServices[string(apfId)] = removeServiceDescription(pos, serviceDescriptions)
 205                         ps.lock.Unlock()
 206                         go ps.sendEvent(*description, eventsapi.CAPIFEventSERVICEAPIUNAVAILABLE)
 207                 }
 208         }
 209         return ctx.NoContent(http.StatusNoContent)
 210 }
 211
 212 // Retrieve a published service API.
 213 func (ps *PublishService) GetApfIdServiceApisServiceApiId(ctx echo.Context, apfId string, serviceApiId string) error {
 214         ps.lock.Lock()
 215         serviceDescriptions, ok := ps.publishedServices[apfId]
 216         ps.lock.Unlock()
 217
 218         if ok {
 219                 _, serviceDescription := getServiceDescription(serviceApiId, serviceDescriptions)
 220                 if serviceDescription == nil {
 221                         return ctx.NoContent(http.StatusNotFound)
 222                 }
 223                 err := ctx.JSON(http.StatusOK, serviceDescription)
 224                 if err != nil {
 225                         // Something really bad happened, tell Echo that our handler failed
 226                         return err
 227                 }
 228
 229                 return nil
 230         }
 231         return ctx.NoContent(http.StatusNotFound)
 232 }
 233
 234 func getServiceDescription(serviceApiId string, descriptions []publishapi.ServiceAPIDescription) (int, *publishapi.ServiceAPIDescription) {
 235         for pos, description := range descriptions {
 236                 // Check for nil as we had a failure here when running unit tests in parallel against a single Capifcore instance
 237                 if (description.ApiId != nil) && (serviceApiId == *description.ApiId) {
 238                         return pos, &description
 239                 }
 240         }
 241         return -1, nil
 242 }
 243
 244 func removeServiceDescription(i int, a []publishapi.ServiceAPIDescription) []publishapi.ServiceAPIDescription {
 245         a[i] = a[len(a)-1]                               // Copy last element to index i.
 246         a[len(a)-1] = publishapi.ServiceAPIDescription{} // Erase last element (write zero value).
 247         a = a[:len(a)-1]                                 // Truncate slice.
 248         return a
 249 }
 250
 251 // Modify an existing published service API.
 252 func (ps *PublishService) ModifyIndAPFPubAPI(ctx echo.Context, apfId string, serviceApiId string) error {
 253         return ctx.NoContent(http.StatusNotImplemented)
 254 }
 255
 256 // Update a published service API.
 257 func (ps *PublishService) PutApfIdServiceApisServiceApiId(ctx echo.Context, apfId string, serviceApiId string) error {
 258         ps.lock.Lock()
 259         defer ps.lock.Unlock()
 260         errMsg := "Unable to update service due to %s."
 261
 262         pos, publishedService, err := ps.checkIfServiceIsPublished(apfId, serviceApiId, ctx)
 263         if err != nil {
 264                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, err))
 265         }
 266
 267         updatedServiceDescription, err := getServiceFromRequest(ctx)
 268         if err != nil {
 269                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, err))
 270         }
 271
 272         // Additional validation for PUT
 273         if (updatedServiceDescription.ApiId == nil) || (*updatedServiceDescription.ApiId != serviceApiId) {
 274                 errDetail := "ServiceAPIDescription ApiId doesn't match path parameter"
 275                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, errDetail))
 276         }
 277
 278         err = ps.checkProfilesRegistered(apfId, *updatedServiceDescription.AefProfiles)
 279         if err != nil {
 280                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, err))
 281         }
 282
 283         ps.updateDescription(pos, apfId, &updatedServiceDescription, &publishedService)
 284
 285         publishedService.AefProfiles = updatedServiceDescription.AefProfiles
 286         ps.publishedServices[apfId][pos] = publishedService
 287
 288         err = ctx.JSON(http.StatusOK, publishedService)
 289         if err != nil {
 290                 // Something really bad happened, tell Echo that our handler failed
 291                 return err
 292         }
 293         return nil
 294 }
 295
 296 func (ps *PublishService) checkIfServiceIsPublished(apfId string, serviceApiId string, ctx echo.Context) (int, publishapi.ServiceAPIDescription, error) {
 297         publishedServices, ok := ps.publishedServices[apfId]
 298         if !ok {
 299                 return 0, publishapi.ServiceAPIDescription{}, fmt.Errorf("service must be published before updating it")
 300         } else {
 301                 for pos, description := range publishedServices {
 302                         if *description.ApiId == serviceApiId {
 303                                 return pos, description, nil
 304                         }
 305                 }
 306         }
 307         return 0, publishapi.ServiceAPIDescription{}, fmt.Errorf("service must be published before updating it")
 308 }
 309
 310 func getServiceFromRequest(ctx echo.Context) (publishapi.ServiceAPIDescription, error) {
 311         var updatedServiceDescription publishapi.ServiceAPIDescription
 312         err := ctx.Bind(&updatedServiceDescription)
 313         if err != nil {
 314                 return publishapi.ServiceAPIDescription{}, fmt.Errorf("invalid format for service")
 315         }
 316         return updatedServiceDescription, nil
 317 }
 318
 319 func (ps *PublishService) updateDescription(pos int, apfId string, updatedServiceDescription, publishedService *publishapi.ServiceAPIDescription) {
 320         if updatedServiceDescription.Description != nil {
 321                 publishedService.Description = updatedServiceDescription.Description
 322                 go ps.sendEvent(*publishedService, eventsapi.CAPIFEventSERVICEAPIUPDATE)
 323         }
 324 }
 325
 326 func (ps *PublishService) sendEvent(service publishapi.ServiceAPIDescription, eventType eventsapi.CAPIFEvent) {
 327         apiIds := []string{*service.ApiId}
 328         apis := []publishapi.ServiceAPIDescription{service}
 329         event := eventsapi.EventNotification{
 330                 EventDetail: &eventsapi.CAPIFEventDetail{
 331                         ApiIds:                 &apiIds,
 332                         ServiceAPIDescriptions: &apis,
 333                 },
 334                 Events: eventType,
 335         }
 336         ps.eventChannel <- event
 337 }
 338
 339 func (ps *PublishService) checkProfilesRegistered(apfId string, updatedProfiles []publishapi.AefProfile) error {
 340         registeredFuncs := ps.serviceRegister.GetAefsForPublisher(apfId)
 341         for _, profile := range updatedProfiles {
 342                 if !slices.Contains(registeredFuncs, profile.AefId) {
 343                         return fmt.Errorf("function %s not registered", profile.AefId)
 344                 }
 345         }
 346         return nil
 347 }
 348
 349 // This function wraps sending of an error in the Error format, and
 350 // handling the failure to marshal that.
 351 func sendCoreError(ctx echo.Context, code int, message string) error {
 352         pd := common29122.ProblemDetails{
 353                 Cause:  &message,
 354                 Status: &code,
 355         }
 356         err := ctx.JSON(code, pd)
 357         return err
 358 }