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         GetAllowedPublishedServices(invokerApiList []publishapi.ServiceAPIDescription) []publishapi.ServiceAPIDescription
  53 }
  54
  55 type PublishService struct {
  56         publishedServices map[string][]publishapi.ServiceAPIDescription
  57         serviceRegister   providermanagement.ServiceRegister
  58         helmManager       helmmanagement.HelmManager
  59         eventChannel      chan<- eventsapi.EventNotification
  60         lock              sync.Mutex
  61 }
  62
  63 // Creates a service that implements both the PublishRegister and the publishserviceapi.ServerInterface interfaces.
  64 func NewPublishService(serviceRegister providermanagement.ServiceRegister, hm helmmanagement.HelmManager, eventChannel chan<- eventsapi.EventNotification) *PublishService {
  65         return &PublishService{
  66                 helmManager:       hm,
  67                 publishedServices: make(map[string][]publishapi.ServiceAPIDescription),
  68                 serviceRegister:   serviceRegister,
  69                 eventChannel:      eventChannel,
  70         }
  71 }
  72
  73 func (ps *PublishService) getAllAefIds() []string {
  74         ps.lock.Lock()
  75         defer ps.lock.Unlock()
  76
  77         allIds := []string{}
  78         for _, descriptions := range ps.publishedServices {
  79                 for _, description := range descriptions {
  80                         allIds = append(allIds, description.GetAefIds()...)
  81                 }
  82         }
  83         return allIds
  84 }
  85
  86 func (ps *PublishService) IsAPIPublished(aefId, path string) bool {
  87         return slices.Contains(ps.getAllAefIds(), aefId)
  88 }
  89
  90 func (ps *PublishService) GetAllPublishedServices() []publishapi.ServiceAPIDescription {
  91         publishedDescriptions := []publishapi.ServiceAPIDescription{}
  92         for _, descriptions := range ps.publishedServices {
  93                 publishedDescriptions = append(publishedDescriptions, descriptions...)
  94         }
  95         return publishedDescriptions
  96 }
  97
  98 func (ps *PublishService) GetAllowedPublishedServices(apiListRequestedServices []publishapi.ServiceAPIDescription) []publishapi.ServiceAPIDescription {
  99         apiListAllPublished := ps.GetAllPublishedServices()
 100         allowedPublishedServices := join(apiListAllPublished, apiListRequestedServices)
 101         return allowedPublishedServices
 102 }
 103
 104 func join(a, b []publishapi.ServiceAPIDescription) []publishapi.ServiceAPIDescription {
 105         var result []publishapi.ServiceAPIDescription
 106
 107         if (a == nil) || (b == nil) || (len(a) == 0) || (len(b) == 0) {
 108                 return result
 109         }
 110
 111         for _, itemA := range a {
 112                 for _, itemB := range b {
 113                         if itemA.ApiName == itemB.ApiName {
 114                                 result = append(result, itemA)
 115                                 break
 116                         }
 117                 }
 118         }
 119         return result
 120 }
 121
 122 // Retrieve all published APIs.
 123 func (ps *PublishService) GetApfIdServiceApis(ctx echo.Context, apfId string) error {
 124         ps.lock.Lock()
 125         serviceDescriptions, ok := ps.publishedServices[apfId]
 126         ps.lock.Unlock()
 127
 128         if ok {
 129                 err := ctx.JSON(http.StatusOK, serviceDescriptions)
 130                 if err != nil {
 131                         // Something really bad happened, tell Echo that our handler failed
 132                         return err
 133                 }
 134         } else {
 135                 if !ps.serviceRegister.IsPublishingFunctionRegistered(apfId) {
 136                         errorMsg := fmt.Sprintf("Unable to get the service due to %s api is only available for publishers", apfId)
 137                         return sendCoreError(ctx, http.StatusNotFound, errorMsg)
 138                 }
 139
 140                 serviceDescriptions = []publishapi.ServiceAPIDescription{}
 141                 err := ctx.JSON(http.StatusOK, serviceDescriptions)
 142                 if err != nil {
 143                         // Something really bad happened, tell Echo that our handler failed
 144                         return err
 145                 }
 146         }
 147         return nil
 148 }
 149
 150 // Publish a new API.
 151 func (ps *PublishService) PostApfIdServiceApis(ctx echo.Context, apfId string) error {
 152         var newServiceAPIDescription publishapi.ServiceAPIDescription
 153         errorMsg := "Unable to publish the service due to %s "
 154         err := ctx.Bind(&newServiceAPIDescription)
 155         if err != nil {
 156                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errorMsg, "invalid format for service "+apfId))
 157         }
 158
 159         if !ps.serviceRegister.IsPublishingFunctionRegistered(apfId) {
 160                 return sendCoreError(ctx, http.StatusForbidden, fmt.Sprintf(errorMsg, "api is only available for publishers "+apfId))
 161         }
 162
 163         if err := ps.isServicePublished(newServiceAPIDescription); err != nil {
 164                 return sendCoreError(ctx, http.StatusForbidden, fmt.Sprintf(errorMsg, err))
 165         }
 166
 167         if err := newServiceAPIDescription.Validate(); err != nil {
 168                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errorMsg, err))
 169         }
 170         ps.lock.Lock()
 171         defer ps.lock.Unlock()
 172
 173         registeredFuncs := ps.serviceRegister.GetAefsForPublisher(apfId)
 174         for _, profile := range *newServiceAPIDescription.AefProfiles {
 175                 if !slices.Contains(registeredFuncs, profile.AefId) {
 176                         return sendCoreError(ctx, http.StatusNotFound, fmt.Sprintf(errorMsg, fmt.Sprintf("function %s not registered", profile.AefId)))
 177                 }
 178         }
 179
 180         newServiceAPIDescription.PrepareNewService()
 181
 182         shouldReturn, returnValue := ps.installHelmChart(newServiceAPIDescription, ctx)
 183         if shouldReturn {
 184                 return returnValue
 185         }
 186         go ps.sendEvent(newServiceAPIDescription, eventsapi.CAPIFEventSERVICEAPIAVAILABLE)
 187
 188         _, ok := ps.publishedServices[apfId]
 189         if ok {
 190                 ps.publishedServices[apfId] = append(ps.publishedServices[apfId], newServiceAPIDescription)
 191         } else {
 192                 ps.publishedServices[apfId] = append([]publishapi.ServiceAPIDescription{}, newServiceAPIDescription)
 193         }
 194
 195         uri := ctx.Request().Host + ctx.Request().URL.String()
 196         ctx.Response().Header().Set(echo.HeaderLocation, ctx.Scheme()+`://`+path.Join(uri, *newServiceAPIDescription.ApiId))
 197         err = ctx.JSON(http.StatusCreated, newServiceAPIDescription)
 198         if err != nil {
 199                 // Something really bad happened, tell Echo that our handler failed
 200                 return err
 201         }
 202
 203         return nil
 204 }
 205
 206 func (ps *PublishService) isServicePublished(newService publishapi.ServiceAPIDescription) error {
 207         for _, services := range ps.publishedServices {
 208                 for _, service := range services {
 209                         if err := service.ValidateAlreadyPublished(newService); err != nil {
 210                                 return err
 211                         }
 212                 }
 213         }
 214         return nil
 215 }
 216
 217 func (ps *PublishService) installHelmChart(newServiceAPIDescription publishapi.ServiceAPIDescription, ctx echo.Context) (bool, error) {
 218         info := strings.Split(*newServiceAPIDescription.Description, ",")
 219         if (len(info) == 5) && (ps.helmManager != nil) {
 220                 err := ps.helmManager.InstallHelmChart(info[1], info[2], info[3], info[4])
 221                 if err != nil {
 222                         return true, sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf("Unable to install Helm chart %s due to: %s", info[3], err.Error()))
 223                 }
 224                 log.Debug("Installed service: ", newServiceAPIDescription.ApiId)
 225         }
 226         return false, nil
 227 }
 228
 229 // Unpublish a published service API.
 230 func (ps *PublishService) DeleteApfIdServiceApisServiceApiId(ctx echo.Context, apfId string, serviceApiId string) error {
 231         serviceDescriptions, ok := ps.publishedServices[string(apfId)]
 232         if ok {
 233                 pos, description := getServiceDescription(serviceApiId, serviceDescriptions)
 234                 if description != nil {
 235                         info := strings.Split(*description.Description, ",")
 236                         if (len(info) == 5) && (ps.helmManager != nil) {
 237                                 ps.helmManager.UninstallHelmChart(info[1], info[3])
 238                                 log.Debug("Deleted service: ", serviceApiId)
 239                         }
 240                         ps.lock.Lock()
 241                         ps.publishedServices[string(apfId)] = removeServiceDescription(pos, serviceDescriptions)
 242                         ps.lock.Unlock()
 243                         go ps.sendEvent(*description, eventsapi.CAPIFEventSERVICEAPIUNAVAILABLE)
 244                 }
 245         }
 246         return ctx.NoContent(http.StatusNoContent)
 247 }
 248
 249 // Retrieve a published service API.
 250 func (ps *PublishService) GetApfIdServiceApisServiceApiId(ctx echo.Context, apfId string, serviceApiId string) error {
 251         ps.lock.Lock()
 252         serviceDescriptions, ok := ps.publishedServices[apfId]
 253         ps.lock.Unlock()
 254
 255         if ok {
 256                 _, serviceDescription := getServiceDescription(serviceApiId, serviceDescriptions)
 257                 if serviceDescription == nil {
 258                         return ctx.NoContent(http.StatusNotFound)
 259                 }
 260                 err := ctx.JSON(http.StatusOK, serviceDescription)
 261                 if err != nil {
 262                         // Something really bad happened, tell Echo that our handler failed
 263                         return err
 264                 }
 265
 266                 return nil
 267         }
 268         return ctx.NoContent(http.StatusNotFound)
 269 }
 270
 271 func getServiceDescription(serviceApiId string, descriptions []publishapi.ServiceAPIDescription) (int, *publishapi.ServiceAPIDescription) {
 272         for pos, description := range descriptions {
 273                 // Check for nil as we had a failure here when running unit tests in parallel against a single Capifcore instance
 274                 if (description.ApiId != nil) && (serviceApiId == *description.ApiId) {
 275                         return pos, &description
 276                 }
 277         }
 278         return -1, nil
 279 }
 280
 281 func removeServiceDescription(i int, a []publishapi.ServiceAPIDescription) []publishapi.ServiceAPIDescription {
 282         a[i] = a[len(a)-1]                               // Copy last element to index i.
 283         a[len(a)-1] = publishapi.ServiceAPIDescription{} // Erase last element (write zero value).
 284         a = a[:len(a)-1]                                 // Truncate slice.
 285         return a
 286 }
 287
 288 // Modify an existing published service API.
 289 func (ps *PublishService) ModifyIndAPFPubAPI(ctx echo.Context, apfId string, serviceApiId string) error {
 290         return ctx.NoContent(http.StatusNotImplemented)
 291 }
 292
 293 // Update a published service API.
 294 func (ps *PublishService) PutApfIdServiceApisServiceApiId(ctx echo.Context, apfId string, serviceApiId string) error {
 295         ps.lock.Lock()
 296         defer ps.lock.Unlock()
 297         errMsg := "Unable to update service due to %s."
 298
 299         pos, publishedService, err := ps.checkIfServiceIsPublished(apfId, serviceApiId)
 300         if err != nil {
 301                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, err))
 302         }
 303
 304         updatedServiceDescription, err := getServiceFromRequest(ctx)
 305         if err != nil {
 306                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, err))
 307         }
 308
 309         // Additional validation for PUT
 310         if (updatedServiceDescription.ApiId == nil) || (*updatedServiceDescription.ApiId != serviceApiId) {
 311                 errDetail := "ServiceAPIDescription ApiId doesn't match path parameter"
 312                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, errDetail))
 313         }
 314
 315         err = ps.checkProfilesRegistered(apfId, *updatedServiceDescription.AefProfiles)
 316         if err != nil {
 317                 return sendCoreError(ctx, http.StatusBadRequest, fmt.Sprintf(errMsg, err))
 318         }
 319
 320         ps.updateDescription(&updatedServiceDescription, &publishedService)
 321
 322         publishedService.AefProfiles = updatedServiceDescription.AefProfiles
 323         ps.publishedServices[apfId][pos] = publishedService
 324
 325         err = ctx.JSON(http.StatusOK, publishedService)
 326         if err != nil {
 327                 // Something really bad happened, tell Echo that our handler failed
 328                 return err
 329         }
 330         return nil
 331 }
 332
 333 func (ps *PublishService) checkIfServiceIsPublished(apfId string, serviceApiId string) (int, publishapi.ServiceAPIDescription, error) {
 334         publishedServices, ok := ps.publishedServices[apfId]
 335         if !ok {
 336                 return 0, publishapi.ServiceAPIDescription{}, fmt.Errorf("service must be published before updating it")
 337         } else {
 338                 for pos, description := range publishedServices {
 339                         if *description.ApiId == serviceApiId {
 340                                 return pos, description, nil
 341                         }
 342                 }
 343         }
 344         return 0, publishapi.ServiceAPIDescription{}, fmt.Errorf("service must be published before updating it")
 345 }
 346
 347 func getServiceFromRequest(ctx echo.Context) (publishapi.ServiceAPIDescription, error) {
 348         var updatedServiceDescription publishapi.ServiceAPIDescription
 349         err := ctx.Bind(&updatedServiceDescription)
 350         if err != nil {
 351                 return publishapi.ServiceAPIDescription{}, fmt.Errorf("invalid format for service")
 352         }
 353         return updatedServiceDescription, nil
 354 }
 355
 356 func (ps *PublishService) updateDescription(updatedServiceDescription, publishedService *publishapi.ServiceAPIDescription) {
 357         if updatedServiceDescription.Description != nil {
 358                 publishedService.Description = updatedServiceDescription.Description
 359                 go ps.sendEvent(*publishedService, eventsapi.CAPIFEventSERVICEAPIUPDATE)
 360         }
 361 }
 362
 363 func (ps *PublishService) sendEvent(service publishapi.ServiceAPIDescription, eventType eventsapi.CAPIFEvent) {
 364         apiIds := []string{*service.ApiId}
 365         apis := []publishapi.ServiceAPIDescription{service}
 366         event := eventsapi.EventNotification{
 367                 EventDetail: &eventsapi.CAPIFEventDetail{
 368                         ApiIds:                 &apiIds,
 369                         ServiceAPIDescriptions: &apis,
 370                 },
 371                 Events: eventType,
 372         }
 373         ps.eventChannel <- event
 374 }
 375
 376 func (ps *PublishService) checkProfilesRegistered(apfId string, updatedProfiles []publishapi.AefProfile) error {
 377         registeredFuncs := ps.serviceRegister.GetAefsForPublisher(apfId)
 378         for _, profile := range updatedProfiles {
 379                 if !slices.Contains(registeredFuncs, profile.AefId) {
 380                         return fmt.Errorf("function %s not registered", profile.AefId)
 381                 }
 382         }
 383         return nil
 384 }
 385
 386 // This function wraps sending of an error in the Error format, and
 387 // handling the failure to marshal that.
 388 func sendCoreError(ctx echo.Context, code int, message string) error {
 389         pd := common29122.ProblemDetails{
 390                 Cause:  &message,
 391                 Status: &code,
 392         }
 393         err := ctx.JSON(code, pd)
 394         return err
 395 }