Enforce memory limits

Summary:
Introducing a whole new exception so we can configure different behavior in ServiceRouter, as well as keeping separate metrics.

This is a crude mechanism for now, but it should be enough to distribute the load better when getting overloaded by a few calls. A followup diff will make the behavior configurable by method and repository.

Reviewed By: clara-9

Differential Revision: D56715749

fbshipit-source-id: 775f52d716c5984132a02c799f1e501440887e85

eden/mononoke/scs/if/source_control.thrift

@@ -2074,6 +2074,11 @@ transient server exception InternalError {
   3: list<string> source_chain;
 }
 
+transient server exception OverloadError {
+  @thrift.ExceptionMessage
+  1: string reason;
+}
+
 struct RequestErrorStruct {
   1: RequestErrorKind kind;
   2: string reason;
@@ -2317,7 +2322,11 @@ service SourceControlService extends fb303_core.BaseService {
   CommitFindFilesResponse commit_find_files(
     1: CommitSpecifier commit,
     2: CommitFindFilesParams params,
-  ) throws (1: RequestError request_error, 2: InternalError internal_error);
+  ) throws (
+    1: RequestError request_error,
+    2: InternalError internal_error,
+    3: OverloadError overload_error,
+  );
 
   CommitHistoryResponse commit_history(
     1: CommitSpecifier commit,

eden/mononoke/scs_server/src/errors.rs

@@ -18,6 +18,7 @@ use source_control_services::errors::source_control_service as service;
 pub(crate) enum ServiceError {
     Request(thrift::RequestError),
     Internal(thrift::InternalError),
+    Overload(thrift::OverloadError),
 }
 
 impl From<thrift::RequestError> for ServiceError {
@@ -32,10 +33,17 @@ impl From<thrift::InternalError> for ServiceError {
     }
 }
 
+impl From<thrift::OverloadError> for ServiceError {
+    fn from(e: thrift::OverloadError) -> Self {
+        Self::Overload(e)
+    }
+}
+
 #[derive(Clone, Copy)]
 pub(crate) enum Status {
     RequestError,
     InternalError,
+    OverloadError,
 }
 
 /// Error can be logged to SCS scuba table
@@ -48,6 +56,7 @@ impl LoggableError for ServiceError {
         match self {
             Self::Request(err) => (Status::RequestError, format!("{:?}", err)),
             Self::Internal(err) => (Status::InternalError, format!("{:?}", err)),
+            Self::Overload(err) => (Status::OverloadError, format!("{:?}", err)),
         }
     }
 }
@@ -77,6 +86,13 @@ impl ServiceError {
                     ..Default::default()
                 })
             }
+            Self::Overload(thrift::OverloadError { reason, .. }) => {
+                let reason = format!("{}: {}", context, reason);
+                Self::Overload(thrift::OverloadError {
+                    reason,
+                    ..Default::default()
+                })
+            }
         }
     }
 }
@@ -214,6 +230,21 @@ macro_rules! impl_into_thrift_error {
                 match e {
                     ServiceError::Request(e) => e.into(),
                     ServiceError::Internal(e) => e.into(),
+                    _ => unreachable!(),
+                }
+            }
+        }
+    };
+}
+
+macro_rules! impl_into_thrift_error_with_oveload {
+    ($t:ty) => {
+        impl From<ServiceError> for $t {
+            fn from(e: ServiceError) -> Self {
+                match e {
+                    ServiceError::Request(e) => e.into(),
+                    ServiceError::Internal(e) => e.into(),
+                    ServiceError::Overload(e) => e.into(),
                 }
             }
         }
@@ -244,7 +275,7 @@ impl_into_thrift_error!(service::CommitInfoExn);
 impl_into_thrift_error!(service::CommitGenerationExn);
 impl_into_thrift_error!(service::CommitCompareExn);
 impl_into_thrift_error!(service::CommitIsAncestorOfExn);
-impl_into_thrift_error!(service::CommitFindFilesExn);
+impl_into_thrift_error_with_oveload!(service::CommitFindFilesExn);
 impl_into_thrift_error!(service::CommitHistoryExn);
 impl_into_thrift_error!(service::CommitListDescendantBookmarksExn);
 impl_into_thrift_error!(service::CommitRunHooksExn);
@@ -382,6 +413,13 @@ pub(crate) fn not_implemented(reason: String) -> thrift::RequestError {
     }
 }
 
+pub(crate) fn overloaded(reason: String) -> thrift::OverloadError {
+    thrift::OverloadError {
+        reason,
+        ..Default::default()
+    }
+}
+
 impl From<GitError> for ServiceError {
     fn from(error: GitError) -> Self {
         match error {

eden/mononoke/scs_server/src/methods/commit.rs

@@ -40,6 +40,7 @@ use mononoke_api::UnifiedDiffMode;
 use mononoke_api::XRepoLookupExactBehaviour;
 use mononoke_api::XRepoLookupSyncBehaviour;
 use mononoke_types::path::MPath;
+use slog::debug;
 use source_control as thrift;
 
 use crate::commit_id::map_commit_identities;
@@ -743,12 +744,50 @@ impl SourceControlServiceImpl {
         commit: thrift::CommitSpecifier,
         params: thrift::CommitFindFilesParams,
     ) -> Result<thrift::CommitFindFilesResponse, errors::ServiceError> {
-        if should_log_memory_usage() {
-            let stats = memory::get_stats();
-            if stats.is_ok() {
-                let mut scuba = ctx.clone().scuba().clone();
-                scuba.add_memory_stats(&stats.unwrap());
-                scuba.log_with_msg("Memory usage before call", None);
+        let rss_min_free_bytes =
+            justknobs::get_as::<usize>("scm/mononoke:scs_rss_min_free_bytes", None).unwrap_or(0);
+        let rss_min_free_pct =
+            justknobs::get_as::<i32>("scm/mononoke:scs_rss_min_free_pct", None).unwrap_or(0);
+
+        debug!(
+            ctx.logger(),
+            "commit_find_files: {} {} {}",
+            rss_min_free_bytes,
+            rss_min_free_pct,
+            should_log_memory_usage(),
+        );
+
+        if rss_min_free_bytes > 0 || rss_min_free_pct > 0 || should_log_memory_usage() {
+            match memory::get_stats() {
+                Ok(stats) => {
+                    debug!(ctx.logger(), "commit_find_files: loaded stats {:?}", stats);
+                    if should_log_memory_usage() {
+                        let mut scuba = ctx.clone().scuba().clone();
+                        scuba.add_memory_stats(&stats);
+                        scuba.log_with_msg("Memory usage before call", None);
+                    }
+                    if stats.rss_free_bytes < rss_min_free_bytes {
+                        debug!(
+                            ctx.logger(),
+                            "not enough memory free, need at least {} bytes free, only {} free right now",
+                            rss_min_free_bytes,
+                            stats.rss_free_bytes,
+                        );
+
+                        return Err(errors::overloaded("Not enough memory free".to_string()).into());
+                    }
+                    if stats.rss_free_pct < rss_min_free_pct as f32 {
+                        debug!(
+                            ctx.logger(),
+                            "not enough memory free, need at least {}% free, only {}% free right now",
+                            rss_min_free_pct,
+                            stats.rss_free_pct,
+                        );
+
+                        return Err(errors::overloaded("Not enough memory free".to_string()).into());
+                    }
+                }
+                Err(_) => {}
             }
         }
 

eden/mononoke/scs_server/src/source_control_impl.rs

@@ -83,6 +83,7 @@ define_stats! {
     total_request_internal_failure: timeseries(Rate, Sum),
     total_request_invalid: timeseries(Rate, Sum),
     total_request_cancelled: timeseries(Rate, Sum),
+    total_request_overloaded: timeseries(Rate, Sum),
 
     // permille is used in canaries, because canaries do not allow for tracking formulas
     total_request_internal_failure_permille: timeseries(Average),
@@ -589,16 +590,17 @@ fn log_result<T: AddScubaResponse>(
 ) {
     let mut scuba = ctx.scuba().clone();
 
-    let (status, error, invalid_request, internal_failure) = match result {
+    let (status, error, invalid_request, internal_failure, overloaded) = match result {
         Ok(response) => {
             response.add_scuba_response(&mut scuba);
-            ("SUCCESS", None, 0, 0)
+            ("SUCCESS", None, 0, 0, 0)
         }
         Err(err) => {
             let (status, desc) = err.status_and_description();
             match status {
-                Status::RequestError => ("REQUEST_ERROR", Some(desc), 1, 0),
-                Status::InternalError => ("INTERNAL_ERROR", Some(desc), 0, 1),
+                Status::RequestError => ("REQUEST_ERROR", Some(desc), 1, 0, 0),
+                Status::InternalError => ("INTERNAL_ERROR", Some(desc), 0, 1, 0),
+                Status::OverloadError => ("OVERLOAD_ERROR", Some(desc), 0, 0, 1),
             }
         }
     };
@@ -610,6 +612,7 @@ fn log_result<T: AddScubaResponse>(
     STATS::total_request_cancelled.add_value(0);
     STATS::total_request_internal_failure_permille.add_value(internal_failure * 1000);
     STATS::total_request_invalid_permille.add_value(invalid_request * 1000);
+    STATS::total_request_overloaded.add_value(overloaded);
 
     ctx.perf_counters().insert_perf_counters(&mut scuba);