Index: /branches/main/array-ingress-controller/apvadapter/annotations/ratelimit/ratelimit.go =================================================================== --- /branches/main/array-ingress-controller/apvadapter/annotations/ratelimit/ratelimit.go (revision 24) +++ /branches/main/array-ingress-controller/apvadapter/annotations/ratelimit/ratelimit.go (working copy) @@ -42,8 +42,8 @@ HardBandwidth int `json:"hard_bandwidth"` MaxReq int `json:"max_req,omitempty"` Timeout int `json:"time_out,omitempty"` - ServerConnReuse bool `json:"server_conn_reuse,omitempty"` - ServerPersistentConn bool `json:"server_persist,omitempty"` + ServerConnReuse bool `json:"server_conn_reuse"` + ServerPersistentConn bool `json:"server_persist"` } func NewRateLimitConfig(serviceName, protocol string, maxCPS, softBW, hardBW, maxReq, timeout int, serverConnReuse, serverPersistentConn bool) *RateLimitConfig { @@ -81,7 +81,7 @@ log.Errorf("Error marshalling rate limit JSON: %v", err) return err } - log.Debugf("Creating Rate limit with Payload: %s", string(jsonData)) + log.Infof("Creating Rate limit with Payload: %s", string(jsonData)) annotationURL := apvurlbuilder.GetAnnotationURL(apvurlbuilder.BaseURL["RealService"], rl.ServiceName) resp, err = client.Update(annotationURL, jsonData) if err != nil { @@ -116,7 +116,7 @@ log.Errorf("Error marshalling rate limit JSON: %v", err) return err } - log.Debugf("Creating Rate limit with Payload: %s", string(jsonData)) + log.Infof("Updating Rate limit with Payload: %s", string(jsonData)) annotationURL := apvurlbuilder.GetAnnotationURL(apvurlbuilder.BaseURL["RealService"], rl.ServiceName) resp, err = client.Update(annotationURL, jsonData) if err != nil { @@ -125,6 +125,7 @@ } log.Debugf("Response Body: %s", resp.String()) log.Debugf("Response Status Code: %v", resp.StatusCode()) + log.Infof("Ratelimit Updated Successfully") return nil } Index: /branches/main/array-ingress-controller/asfadapter/cmd.go =================================================================== --- /branches/main/array-ingress-controller/asfadapter/cmd.go (revision 24) +++ /branches/main/array-ingress-controller/asfadapter/cmd.go (working copy) @@ -45,25 +45,13 @@ return services } -<<<<<<< .mine //fetchAsfHealthChecks fetches all health checks associated with the given service name func fetchAsfHealthChecks() []configstore.AsfHealthCheck { log.Info("Fetching ASF Healthchecks...") var healthchecks []configstore.AsfHealthCheck -||||||| .r12 -// processService processes each AsfRealService -func processService(svc *configstore.AsfRealService) { -======= -//fetchAsfHealthChecks fetches all health checks associated with the given service name -func fetchAsfHealthChecks() []configstore.AsfHealthCheck { - log.Info("Fetching Healthchecks...") - - var healthchecks []configstore.AsfHealthCheck ->>>>>>> .r19 dbclient, err := configstore.Get() if dbclient == nil { log.Error("DB Client is NIL") -<<<<<<< .mine return nil } @@ -81,56 +69,23 @@ dbclient, err := configstore.Get() if dbclient == nil { log.Error("DB Client is NIL") -||||||| .r12 -======= - return nil - } - - // Fetch only healthchecks with status = 'WAITING' - err = dbclient.ReadByStatus("WAITING", &healthchecks) - if err != nil { - log.Errorf("Error fetching healthchecks: %v", err) - return nil - } - - return healthchecks -} - -// ProcessAsfService processes each AsfRealService -func ProcessAsfService(svc *configstore.AsfRealService) { - dbclient, err := configstore.Get() - if dbclient == nil { - log.Error("DB Client is NIL") ->>>>>>> .r19 return } const MaxRetries = 5 retryCount := 0 -<<<<<<< .mine // Check if ASF is available // If TEST environment is set, assume ASF is available // Otherwise, check the status from the watchtower if os.Getenv("TEST") == "true" { log.Info("TEST environment detected, setting APV availability to true") available := true -||||||| .r12 - - select { - case available := <-watchtower.GetDeviceStatusChan(): -======= - - if os.Getenv("TEST") == "true" { - log.Info("TEST environment detected, setting APV availability to true") - available := true ->>>>>>> .r19 if !available { log.Error("ASF is not available") return } else { log.Info("ASF is available") } -<<<<<<< .mine } else { select { case available := <-watchtower.GetDeviceStatusChan(): @@ -144,25 +99,6 @@ log.Warn("ASF status not received") return } -||||||| .r12 - case <-time.After(30 * time.Second): - log.Warn("ASF status not received") - return -======= - } else { - select { - case available := <-watchtower.GetDeviceStatusChan(): - if !available { - log.Error("ASF is not available") - return - } else { - log.Info("ASF is available") - } - case <-time.After(30 * time.Second): - log.Warn("ASF status not received") - return - } ->>>>>>> .r19 } // Update status to INPROGRESS @@ -214,15 +150,8 @@ return } -<<<<<<< .mine // Log the error and retry log.Errorf("Error processing ASF RealService %s (attempt %d/%d): %v", svc.ServiceName, retryCount+1, MaxRetries, err) -||||||| .r12 - log.Errorf("Error processing service %s (attempt %d/%d): %v", svc.ServiceName, retryCount+1, maxRetries, err) -======= - // Log the error and retry - log.Errorf("Error processing service %s (attempt %d/%d): %v", svc.ServiceName, retryCount+1, MaxRetries, err) ->>>>>>> .r19 retryCount++ time.Sleep(5 * time.Second) } @@ -237,7 +166,6 @@ } } -<<<<<<< .mine func ProcessAsfHealthCheck(HChecks *configstore.AsfHealthCheck) { dbclient, err := configstore.Get() if dbclient == nil { @@ -343,110 +271,6 @@ } -||||||| .r12 -======= -func ProcessAsfHealthCheck(HChecks *configstore.AsfHealthCheck) { - dbclient, err := configstore.Get() - if dbclient == nil { - log.Error("DB Client is NIL") - return - } - - const MaxRetries = 5 - retryCount := 0 - - if os.Getenv("TEST") == "true" { - log.Info("TEST environment detected, setting APV availability to true") - available := true - if !available { - log.Error("ASF is not available") - return - } else { - log.Info("ASF is available") - } - } else { - select { - case available := <-watchtower.GetDeviceStatusChan(): - if !available { - log.Error("ASF is not available") - return - } else { - log.Info("ASF is available") - } - case <-time.After(30 * time.Second): - log.Warn("ASF status not received") - return - } - } - - // Update status to INPROGRESS - HChecks.Status = "INPROGRESS" - conditions := map[string]interface{}{ - "service_name": HChecks.ServiceName, - // Add more conditions as needed - } - if err = dbclient.Update(HChecks, conditions); err != nil { - log.Errorf("Failed to update status to INPROGRESS for healthcheck of service %s: %v", HChecks.ServiceName, err) - return - } - - hc := health.NewAsfHealthCheck( - HChecks.ServiceName, - HChecks.ServiceType, - HChecks.CheckType, - HChecks.IPv4, - HChecks.Port, - HChecks.HCUp, - HChecks.HCDown, - ) - - // Retry loop for create, update, or delete - for retryCount < MaxRetries { - switch HChecks.CurrentOp { - case "CREATE": - err = hc.CreateHealthCheck() - case "UPDATE": - err = hc.UpdateHealthCheck() - case "DELETE": - err = hc.DeleteHealthCheck() - default: - log.Errorf("Invalid operation %s for healthcheck on service %s", HChecks.CurrentOp, HChecks.ServiceName) - HChecks.Status = "FAILED" - _ = dbclient.Update(HChecks, conditions) - return - } - - if err == nil { - // Success → Update status to COMPLETED - HChecks.Status = "COMPLETED" - if err := dbclient.Update(HChecks, conditions); err != nil { - log.Errorf("Failed to update status to COMPLETED for healthcheck of service %s: %v", HChecks.ServiceName, err) - } - if HChecks.CurrentOp == "DELETE" { - //Remove HealthCheck entry after deletion process - removeAssociatedHealthChecks(HChecks.ServiceName) - log.Infof("Deleted HealthCheck record after successful deletion %s", HChecks.ServiceName) - } - return - } - - // Log the error and retry - log.Errorf("Error processing healthcheck of service %s (attempt %d/%d): %v", HChecks.ServiceName, retryCount+1, MaxRetries, err) - retryCount++ - time.Sleep(5 * time.Second) - } - - // If we reach max retries → Mark as FAILED - HChecks.Status = "FAILED" - if err := dbclient.Update(HChecks, conditions); err != nil { - log.Errorf("Failed to update status to FAILED for healthcheck of service %s: %v", HChecks.ServiceName, err) - } else { - log.Errorf("Healthcheck of %s failed after %d retries, marked as FAILED", HChecks.ServiceName, MaxRetries) - } - -} - ->>>>>>> .r19 // StartReconcile continuously processes services in the queue // This function runs in a loop, fetching services and health checks from the database // and processing them @@ -455,15 +279,8 @@ func StartReconcile() { for { log.Info("Reconciling....") -<<<<<<< .mine svcs := FetchAsfServices() // Check if there are any services to process -||||||| .r12 - svcs := fetchAsfServices() - -======= - svcs := FetchAsfServices() ->>>>>>> .r19 if len(svcs) == 0 { log.Info("No AsfRealServices in DB to process") }else { @@ -490,50 +307,15 @@ } } -<<<<<<< .mine //UTILS functiions -||||||| .r12 - var wg sync.WaitGroup - for i := range svcs { - wg.Add(1) - go func(svc *configstore.AsfRealService) { - defer wg.Done() - log.Infof("Processing AsfRealService: %s", svc.ServiceName) - processService(svc) - }(&svcs[i]) - } -======= - hcs := fetchAsfHealthChecks() - if len(hcs) == 0 { - log.Info("No HealthChecks") - } else { - for i := range hcs { - hc := &hcs[i] - log.Infof("Processing HealthCheck of service: %s", hc.ServiceName) - ProcessAsfHealthCheck(hc) - } - } ->>>>>>> .r19 - - time.Sleep(120* time.Second) - -<<<<<<< .mine // removeAssociatedHealthChecks deletes all health checks associated with the given service name func removeAssociatedHealthChecks(serviceName string) { dbclient, err := configstore.Get() if err != nil { log.Errorf("Error getting DB client: %v", err) return -||||||| .r12 - wg.Wait() - time.Sleep(120 * time.Second) -======= - //fetch all the health checks associated with the service - - //process all the healthchecks ->>>>>>> .r19 } if dbclient == nil { log.Error("DB Client is NIL") @@ -554,7 +336,6 @@ log.Infof("Successfully removed healthChecks for ASF RealService %s", serviceName) } } -<<<<<<< .mine // removeRealService deletes the real service associated with the given service name func removeRealService(serviceName string) { @@ -608,85 +389,3 @@ } } -||||||| .r12 -======= - - -//UTILS functiions - -// removeAssociatedHealthChecks deletes all health checks associated with the given service name -func removeAssociatedHealthChecks(serviceName string) { - dbclient, err := configstore.Get() - if err != nil { - log.Errorf("Error getting DB client: %v", err) - return - } - if dbclient == nil { - log.Error("DB Client is NIL") - return - } - - log.Infof("Removing health checks associated with service %s", serviceName) - - conditions := map[string]interface{}{ - "service_name": serviceName, - } - - if deleteErr := dbclient.Delete(&configstore.AsfHealthCheck{}, conditions); deleteErr != nil { - log.Errorf("Failed to remove health checks for service %s: %v", serviceName, deleteErr) - } else { - log.Infof("Successfully removed health checks for service %s", serviceName) - } -} - -// removeRealService deletes the real service associated with the given service name -func removeRealService(serviceName string) { - dbclient, err := configstore.Get() - if err != nil { - log.Errorf("Error getting DB client: %v", err) - return - } - if dbclient == nil { - log.Error("DB Client is NIL") - return - } - - log.Infof("Removing real service associated with service %s", serviceName) - - conditions := map[string]interface{}{ - "service_name": serviceName, - } - - if deleteErr := dbclient.Delete(&configstore.AsfRealService{}, conditions); deleteErr != nil { - log.Errorf("Failed to remove real service for service %s: %v", serviceName, deleteErr) - } else { - log.Infof("Successfully removed real service for service %s", serviceName) - } -} - -// MarkLinkedResourcesAsFailed marks all associated health checks as FAILED -func MarkLinkedResourcesAsFailed(serviceName string) { - dbclient, err := configstore.Get() - if dbclient == nil { - log.Error("DB Client is NIL") - return - } - - // Fetch and update linked HealthChecks - var healthChecks []configstore.AsfHealthCheck - err = dbclient.ReadByCondition(map[string]interface{}{"service_name": serviceName}, &healthChecks) - if err != nil { - log.Errorf("Failed to fetch linked HealthChecks for service %s: %v", serviceName, err) - } else { - for _, hc := range healthChecks { - hc.Status = "FAILED" - if err := dbclient.Update(&hc, map[string]interface{}{"hc_name": hc.ServiceName}); err != nil { - log.Errorf("Failed to update status to FAILED for HealthCheck of service %s: %v", hc.ServiceName, err) - } else { - log.Infof("Marked HealthCheck as FAILED due to service %s failure", hc.ServiceName) - } - } - } -} - ->>>>>>> .r19 Index: /branches/main/array-ingress-controller/asfadapter/realservice/realservice.go =================================================================== --- /branches/main/array-ingress-controller/asfadapter/realservice/realservice.go (revision 24) +++ /branches/main/array-ingress-controller/asfadapter/realservice/realservice.go (working copy) @@ -118,25 +118,16 @@ } log.Infof("Generated payload: %v", string(jsonData)) -<<<<<<< .mine // Build the URL for the ASF API // The URL is built using the GetAnnotationURL function from the asfurlbuilder package annotationUrl := asfurlbuilder.GetAnnotationURL(asfurlbuilder.BaseURL["RealService"]) resp, err = client.Create(annotationUrl, jsonData) -||||||| .r12 - annotationUrl := asfurlbuilder.GetAnnotationURL(asfurlbuilder.BaseURL["RealService"], "") - _, err = client.Create(annotationUrl, jsonData) -======= - annotationUrl := asfurlbuilder.GetAnnotationURL(asfurlbuilder.BaseURL["RealService"]) - resp, err = client.Create(annotationUrl, jsonData) ->>>>>>> .r19 if err != nil { log.Errorf("Error creating real service: %v", err) return err } -<<<<<<< .mine log.Debugf("Successfully created real service: %s", rs.ServiceName) log.Debugf("Response Body: %s", resp.String()) @@ -148,23 +139,6 @@ return err } log.Infof("Response Status Code: %v", resp.StatusCode()) -||||||| .r12 - log.Infof("Successfully created real service: %s", rs.ServiceName) -======= - log.Debugf("url: %s", annotationUrl) - log.Infof("Request Payload: %s", string(jsonData)) - log.Infof("Successfully created real service: %s", rs.ServiceName) ->>>>>>> .r19 - log.Infof("Response Body: %s", resp.String()) - - //Since ASF returns status 200 for even incorrect requests - //we need to check the response body for the word "Error" - //if the word "Error" is present in resp.String() return error - if regexp.MustCompile(`(?i)Error`).FindString(resp.String()) != "" { - log.Errorf("Error in response: %s", resp.String()) - return err - } - log.Infof("Response Status Code: %v", resp.StatusCode()) return nil } @@ -178,7 +152,6 @@ } log.Infof("Generated payload: %v", string(jsonData)) -<<<<<<< .mine // Build the URL for the ASF API // The URL is built using the GetAnnotationURL function from the asfurlbuilder package annotationUrl := asfurlbuilder.GetAnnotationURL(asfurlbuilder.BaseURL["RealService"]) @@ -186,21 +159,11 @@ //Create function is used because POST method has been used for updating the ASF real service //This has been done bbecause we are using CLI based commands to update the real service instead of actual API call resp, err = client.Create(annotationUrl, jsonData) -||||||| .r12 - annotationUrl := asfurlbuilder.GetAnnotationURL(asfurlbuilder.BaseURL["RealService"], rs.ServiceName) - _, err = client.Update(annotationUrl, jsonData) -======= - annotationUrl := asfurlbuilder.GetAnnotationURL(asfurlbuilder.BaseURL["RealService"]) - //Create function is used because POST method has been used for updating the ASF real service - //This has been done bbecause we are using CLI based commands to update the real service instead of actual API call - resp, err = client.Create(annotationUrl, jsonData) ->>>>>>> .r19 if err != nil { log.Errorf("Error updating real service: %v", err) return err } - log.Debugf("url: %s", annotationUrl) log.Infof("Successfully updated real service: %s", rs.ServiceName) log.Infof("Response Body: %s", resp.String()) log.Infof("Response Status Code: %v", resp.StatusCode()) @@ -226,7 +189,6 @@ return err } - log.Debugf("url: %s", annotationUrl) log.Infof("Successfully deleted real service: %s", rs.ServiceName) log.Infof("Response Body: %s", resp.String()) log.Infof("Response Status Code: %v", resp.StatusCode()) Index: /branches/main/array-ingress-controller/configstore/models.go =================================================================== --- /branches/main/array-ingress-controller/configstore/models.go (revision 24) +++ /branches/main/array-ingress-controller/configstore/models.go (working copy) @@ -75,7 +75,6 @@ ServiceType string `gorm:"column:service_type;not null"` IPv4 string `gorm:"column:ipv4;not null"` Port int `gorm:"column:port;not null"` -<<<<<<< .mine Status string `gorm:"column:status;not null"` CurrentOp string `gorm:"column:current_op;not null"` } @@ -88,21 +87,6 @@ IPv4 string `gorm:"column:ipv4;not null"` Port int `gorm:"column:port;not null"` CheckType string `gorm:"column:hc_type;not null"` -||||||| .r12 - CheckType string `gorm:"column:check_type;not null"` -======= - Status string `gorm:"column:status;not null"` - CurrentOp string `gorm:"column:current_op;not null"` -} - -type AsfHealthCheck struct { - IngressName string `gorm:"column:ingress_name;not null"` - ServiceName string `gorm:"column:service_name;not null"` - ServiceType string `gorm:"column:service_type;not null"` - IPv4 string `gorm:"column:ipv4;not null"` - Port int `gorm:"column:port;not null"` - CheckType string `gorm:"column:hc_type;not null"` ->>>>>>> .r19 HCUp int `gorm:"column:hc_up;not null"` HCDown int `gorm:"column:hc_down;not null"` Status string `gorm:"column:status;not null"` Index: /branches/main/array-ingress-controller/controller/kubernetes-controller/handlers/ingress_event_handler.go =================================================================== --- /branches/main/array-ingress-controller/controller/kubernetes-controller/handlers/ingress_event_handler.go (revision 24) +++ /branches/main/array-ingress-controller/controller/kubernetes-controller/handlers/ingress_event_handler.go (working copy) @@ -209,16 +209,14 @@ return } - log.Infof("Ingress deleted: %s/%s\n", ingress.Namespace, ingress.Name) + log.Infof("Ingress to be deleted: %s/%s\n", ingress.Namespace, ingress.Name) - // Extract associated service name from ingress - assocSvc, err := utils.GetAssocSvc(ingress.Namespace, ingress.Name) + assocSvc, err := utils.GetAssocSvcDel(ingress) if err != nil { log.Errorf("Ingress '%s' in namespace '%s' has no associated service", ingress.Name, ingress.Namespace) return } log.Infof("Associated service: %s", assocSvc) - // Get endpoint using service name // Endpoints may not exist if service was deleted or pods are gone - this is OK for deletion var endpoints []*corev1.Endpoints Index: /branches/main/array-ingress-controller/controller/kubernetes-controller/utils/k8s_helper.go =================================================================== --- /branches/main/array-ingress-controller/controller/kubernetes-controller/utils/k8s_helper.go (revision 24) +++ /branches/main/array-ingress-controller/controller/kubernetes-controller/utils/k8s_helper.go (working copy) @@ -81,6 +81,7 @@ log.Errorf("Annotation '%s' does not exist.\n", annotationKey) return "", exists } + log.Infof("ANNOT : %s. AnnotValue : %s\n", annotationKey, jsonString) return jsonString, exists } @@ -248,6 +249,7 @@ } // GetAssocSvc extracts service name from ingress + func GetAssocSvc(namespace, ingressName string) (string, error) { ingress, err := KubeClient.NetworkingV1().Ingresses(namespace).Get(context.TODO(), ingressName, metav1.GetOptions{}) if err != nil { @@ -272,6 +274,37 @@ return path.Backend.Service.Name, nil } +func GetAssocSvcDel(ing *networkingv1.Ingress) (string, error) { + if ing == nil { + return "", errors.New("no ingress found") + } + + // Check default backend first + if ing.Spec.DefaultBackend != nil && ing.Spec.DefaultBackend.Service != nil { + if name := ing.Spec.DefaultBackend.Service.Name; name != "" { + return name, nil + } + } + + // Check rules → http → paths + for _, rule := range ing.Spec.Rules { + if rule.HTTP == nil { + continue + } + for _, p := range rule.HTTP.Paths { + if p.Backend.Service != nil { + if name := p.Backend.Service.Name; name != "" { + return name, nil + } + } + } + } + + // No service found + return "", errors.New("no service backend found") +} + + // GetServicefromName retrieves a service by name func GetServicefromName(namespace, serviceName string) (*corev1.Service, error) { service, err := KubeClient.CoreV1().Services(namespace).Get(context.TODO(), serviceName, metav1.GetOptions{}) Index: /branches/main/array-ingress-controller/task-manager/kubernetes/apv/health.go =================================================================== --- /branches/main/array-ingress-controller/task-manager/kubernetes/apv/health.go (revision 24) +++ /branches/main/array-ingress-controller/task-manager/kubernetes/apv/health.go (working copy) @@ -20,6 +20,12 @@ ExpectedCode *string `json:"expected Code,omitempty"` Interval *int `json:"interval,omitempty"` Timeout *int `json:"timeout,omitempty"` + + IPv4 *string `json:"ipv4,omitempty"` + Port *int `json:"port,omitempty"` + HcUp *int `json:"hc_up,omitempty"` + HcDown *int `json:"hc_down,omitempty"` + Retries *int `json:"retries,omitempty"` } @@ -74,10 +80,11 @@ HCName: hcName, ServiceName: realServiceName, Type: getStringValue(healthCheckAnnot.Type, ""), - HcUp: DefaultHcUp, - HcDown: DefaultHcDown, - IPv4: address.IP, - Port: port, + HcUp: getIntValue(healthCheckAnnot.HcUp, DefaultHcUp), + HcDown: getIntValue(healthCheckAnnot.HcDown, DefaultHcDown), + IPv4: getStringValue(healthCheckAnnot.IPv4, address.IP), + Port: getIntValue(healthCheckAnnot.Port, port), + SendInterval: getIntValue(healthCheckAnnot.Interval, 3), ServerTimeout: getIntValue(healthCheckAnnot.Timeout, 5), Retries: getIntValue(healthCheckAnnot.Retries, 3), @@ -90,7 +97,8 @@ return err } - log.Infof("Health check record created successfully for pod %s", realServiceName) + log.Infof("K8's Health check record created successfully for pod %s", realServiceName) + log.Infof("K8's Health check record HC name : %s\n", hcName); } } @@ -133,10 +141,11 @@ HCName: "hc" + realService.ServiceName, ServiceName: realService.ServiceName, Type: getStringValue(healthCheckAnnot.Type, ""), - HcUp: DefaultHcUp, - HcDown: DefaultHcDown, - IPv4: realService.IPAddress, - Port: realService.Port, + HcUp: getIntValue(healthCheckAnnot.HcUp, DefaultHcUp), + HcDown: getIntValue(healthCheckAnnot.HcDown, DefaultHcDown), + IPv4: getStringValue(healthCheckAnnot.IPv4, realService.IPAddress), + Port: getIntValue(healthCheckAnnot.Port, realService.Port), + SendInterval: getIntValue(healthCheckAnnot.Interval, 3), ServerTimeout: getIntValue(healthCheckAnnot.Timeout, 5), Retries: getIntValue(healthCheckAnnot.Retries, 3), Index: /branches/main/array-ingress-controller/task-manager/kubernetes/apv/realservice.go =================================================================== --- /branches/main/array-ingress-controller/task-manager/kubernetes/apv/realservice.go (revision 24) +++ /branches/main/array-ingress-controller/task-manager/kubernetes/apv/realservice.go (working copy) @@ -162,6 +162,7 @@ log.Errorf("Error getting port from ingress %s: %v", task.Ingress[0].Name, err) return err } + log.Infof("Update real port : %d\n", port); // Get protocol from ingress protocol := utils.GetIngressProtocol(task.Ingress[0]) Index: /branches/main/array-ingress-controller/task-manager/openshift/asf/const.go =================================================================== --- /branches/main/array-ingress-controller/task-manager/openshift/asf/const.go (revision 24) +++ /branches/main/array-ingress-controller/task-manager/openshift/asf/const.go (working copy) @@ -21,28 +21,16 @@ DefaultHcUp = 3 // this is default value set at ASF's end when no hc up count is passed DefaultHcDown = 3 // this is default value set at ASF's end when no hc down count is passed -<<<<<<< .mine // annotations keys expected to be defined in route.yaml etc. HealthCheckAnnotKey = "asf.arraynetworks.com/healthcheck" ASFRealServiceAnnotKey = "asf.arraynetworks.com/asf-realservice" -||||||| .r11 - AsfRealServiceAnnotKey = "arraynetworks.com/asfrealservice" - DefaultCheckType = "http" -======= - // annotations keys expected to be defined in route.yaml etc. - HealthCheckAnnotKey = "asf.arraynetworks.com/healthcheck" ->>>>>>> .r19 ) // Function map to handle different task/resource combinations var asfFunctionMap = map[string]SetUpFunc{ - "ROUTE_ADD": handleRouteAdd, - "ROUTE_UPDATE": handleRouteUpdate, - "ROUTE_DELETE": handleRouteDelete, - // TODO: Uncomment once pod handlers are ready. - // "POD_ADD": handlePodAdd, - // "POD_DELETE": handlePodDelete, - // "SERVICE_UPDATE": handleServiceUpdate, + "ROUTE_ADD": handleRouteAdd, + "ROUTE_UPDATE": handleRouteUpdate, + "ROUTE_DELETE": handleRouteDelete, "ENDPOINT_ADD": handleEndpointAdd, "ENDPOINT_DELETE": handleEndpointDelete, } Index: /branches/main/array-ingress-controller/task-manager/openshift/asf/realservice.go =================================================================== --- /branches/main/array-ingress-controller/task-manager/openshift/asf/realservice.go (revision 24) +++ /branches/main/array-ingress-controller/task-manager/openshift/asf/realservice.go (working copy) @@ -1,6 +1,18 @@ +/* + + Copyright 2025 Array Networks + + This file manages the controller-side operations related to RealService + functionalities.It includes functions to add, update, and delete realservice configurations in db + based on route. Here we are just marking the rows in with intended Status and Operation + for a specific healthcheck in different scenarios. Actual CRUD is performed in ASF Adapter module. + +*/ + package asf import ( + "encoding/json" "fmt" "arraynetworks.com/array-ingress-controller/configstore" @@ -8,6 +20,9 @@ taskManager "arraynetworks.com/array-ingress-controller/task-manager/task" ) +// addRealService adds a realservice configuration to the config store +// based on the annotation present on the route. It initializes new realservice +// records for each real service associated with the route. func addRealService(task *taskManager.Task) error { log.Info("Handling Route Add with additional context data") @@ -17,7 +32,6 @@ log.Error("Error initializing config store: %v", err) return err } -<<<<<<< .mine if task.EndPts == nil || len(task.EndPts) == 0 || task.EndPts[0].Subsets == nil { log.Warnf("No endpoint subsets found for route %s", task.Route[0].Name) return nil @@ -32,35 +46,13 @@ } else { serviceType = parsed["service_type"] } -||||||| .r11 - - annots, exists := utils.FetchJSONStringFromAnnotation(task.Route[0], AsfRealServiceAnnotKey) - - if !exists { - logger.Warnf("No ASF annotations found for route %s", task.Route[0].Name) -======= - if task.EndPts == nil || len(task.EndPts) == 0 || task.EndPts[0].Subsets == nil { - log.Warnf("No endpoint subsets found for route %s", task.Route[0].Name) - return nil ->>>>>>> .r19 } -<<<<<<< .mine // Fallback to inferred protocol if not provided explicitly if serviceType == "" { serviceType = utils.GetRouteProtocol(task.Route[0]) } -||||||| .r11 - // Unmarshal the JSON string into a HealthCheck struct - var asfRealServAnnot AsfRealServiceAnnot - err = json.Unmarshal([]byte(annots), &asfRealServAnnot) - if err != nil { - return fmt.Errorf("error unmarshaling JSON from annotation: %v", err) - } - -======= ->>>>>>> .r19 // Create a service entry for each endpoint address for _, subset := range task.EndPts[0].Subsets { for _, address := range subset.Addresses { @@ -82,28 +74,14 @@ realService := configstore.AsfRealService{ IngressName: task.Route[0].Name, ServiceName: realServiceName, -<<<<<<< .mine ServiceType: serviceType, -||||||| .r11 - ServiceType: getStringValue(asfRealServAnnot.ServiceType, DefaultServiceType), -======= - ServiceType: utils.GetServiceProtocolForRoute(task.Route[0]), ->>>>>>> .r19 IPv4: address.IP, Port: int(task.Route[0].Spec.Port.TargetPort.IntValue()), Status: StatusWaiting, CurrentOp: OperationCreate, } -<<<<<<< .mine if err = configStore.Create(&realService); err != nil { log.Errorf("Error creating service record for pod %s: %v", realServiceName, err) -||||||| .r11 - if err = configStore.Create(realService); err != nil { - logger.Errorf("Error creating service record for pod %s: %v", realServiceName, err) -======= - if err = configStore.Create(realService); err != nil { - log.Errorf("Error creating service record for pod %s: %v", realServiceName, err) ->>>>>>> .r19 return err } else { log.Infof("Service record created successfully for pod %s", realServiceName) @@ -113,6 +91,7 @@ return nil } +// deleteRealService deletes existing realservice configurations in the config store. func deleteRealService(task *taskManager.Task) error { // Initialize the config store configStore, err := configstore.Get() @@ -176,6 +155,8 @@ return nil } +// updateRealService updates existing realservice configurations in the config store +// based on the changes in route spec func updateRealService(task *taskManager.Task) error { // Initialize the config store configStore, err := configstore.Get()