func()

in api/query_handler.go [107:216]
95 lines of code
14 McCabe index (conditional complexity)

func (handler *QueryHandler) handleAQLInternal(aqlRequest apiCom.AQLRequest, w *utils.ResponseWriter, r *http.Request) {
	var err error
	var duration time.Duration
	var qcs []*query.AQLQueryContext
	var statusCode int

	defer func() {
		var errStr string
		if err != nil {
			errStr = err.Error()
		}

		l := utils.GetQueryLogger().With(
			"error", errStr,
			"request", aqlRequest,
			"queries_enabled_", aqlRequest.Body.Queries,
			"duration", duration,
			"statusCode", statusCode,
			"contexts_enabled_", qcs,
			"headers", r.Header,
		)

		if statusCode == http.StatusOK {
			l.Info("All queries succeeded")
		} else {
			l.Error("Some of the queries finished with error")
		}
	}()

	if aqlRequest.Query != "" {
		// Override from query parameter
		err = json.Unmarshal([]byte(aqlRequest.Query), &aqlRequest.Body)
		if err != nil {
			statusCode = http.StatusBadRequest
			w.WriteError(utils.APIError{
				Code:    statusCode,
				Message: ErrMsgFailedToUnmarshalRequest,
				Cause:   err,
			})
			return
		}
	}

	if aqlRequest.Body.Queries == nil {
		statusCode = http.StatusBadRequest
		w.WriteError(ErrMissingParameter)
		return
	}

	returnHLL := aqlRequest.Accept == utils.HTTPContentTypeHyperLogLog
	if aqlRequest.DeviceChoosingTimeout <= 0 {
		aqlRequest.DeviceChoosingTimeout = -1
	}

	queryTimer := utils.GetRootReporter().GetTimer(utils.QueryLatency)
	start := utils.Now()
	var requestResponseWriter QueryResponseWriter

	if !returnHLL && canEagerFlush(aqlRequest.Body.Queries) {
		statusCode = http.StatusOK
		aqlQuery := aqlRequest.Body.Queries[0]
		qc := &query.AQLQueryContext{
			Query:         &aqlQuery,
			ReturnHLLData: false,
			DataOnly:      aqlRequest.DataOnly != 0,
		}
		qc.InitQCHelper()
		qc.Compile(handler.memStore, handler.shardOwner)
		qc.ResponseWriter = w
		if qc.Error != nil {
			err = qc.Error
			statusCode = http.StatusBadRequest
			w.WriteErrorWithCode(statusCode, err)
			return
		}
		// for logging purpose only
		qcs = append(qcs, qc)

		qc.FindDeviceForQuery(handler.memStore, aqlRequest.Device, handler.deviceManager, aqlRequest.DeviceChoosingTimeout)
		if qc.Error != nil {
			err = qc.Error
			statusCode = http.StatusServiceUnavailable
			w.WriteErrorWithCode(statusCode, err)
			return
		}
		defer handler.deviceManager.ReleaseReservedMemory(qc.Device, qc.Query)

		qc.ProcessQuery(handler.memStore)
		if qc.Error != nil {
			err = qc.Error
			utils.GetQueryLogger().With(
				"error", err,
				"query", aqlQuery,
				"context", qc,
			).Error("Error happened when processing query")
			statusCode = http.StatusInternalServerError
			return
		}

		if !qc.DataOnly {
			w.Write([]byte(`]}]`))

			if aqlRequest.Verbose > 0 {
				w.Write([]byte(`,"context":`))
				qcBytes, _ := json.Marshal(qcs)
				w.Write(qcBytes)
			}

			w.Write([]byte(`}`))
		}