capifcore/internal/publishservice/publishservice.go

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