Method CreateWorkload

pkg/async-gateway/service.go:57–92

CreateWorkload enqueues an async workload request and uploads the request payload to S3

(id string, apiName string, queueURL string, payload io.Reader, headers http.Header)

// CreateWorkload enqueues an async workload request and uploads the request payload to S3
func (s *service) CreateWorkload(id string, apiName string, queueURL string, payload io.Reader, headers http.Header) (string, error) {
	prefix := async.StoragePath(s.clusterUID, apiName)
	log := s.logger.With(zap.String("id", id), zap.String("apiName", apiName))

	buf := &bytes.Buffer{}
	if err := json.NewEncoder(buf).Encode(headers); err != nil {
		return "", errors.Wrap(err, "failed to dump headers")
	}

	headersPath := async.HeadersPath(prefix, id)
	log.Debugw("uploading headers", zap.String("path", headersPath))
	if err := s.storage.Upload(headersPath, buf, "application/json"); err != nil {
		return "", errors.Wrap(err, "failed to upload headers")
	}

	contentType := headers.Get("Content-Type")
	payloadPath := async.PayloadPath(prefix, id)
	log.Debugw("uploading payload", zap.String("path", payloadPath))
	if err := s.storage.Upload(payloadPath, payload, contentType); err != nil {
		return "", errors.Wrap(err, "failed to upload payload")
	}

	log.Debug("sending message to queue")
	queue := NewSQS(queueURL, &s.session)
	if err := queue.SendMessage(id, id); err != nil {
		return "", errors.Wrap(err, "failed to send message to queue")
	}

	statusPath := fmt.Sprintf("%s/%s/status/%s", prefix, id, async.StatusInQueue)
	log.Debug(fmt.Sprintf("setting status to %s", async.StatusInQueue))
	if err := s.storage.Upload(statusPath, strings.NewReader(""), "text/plain"); err != nil {
		return "", errors.Wrap(err, "failed to upload workload status")
	}

	return id, nil
}

// GetWorkload retrieves the status and result, if available, of a given workload
func (s *service) GetWorkload(id string, apiName string) (GetWorkloadResponse, error) {